Charles Leung
03/16/2021, 3:02 PMValueError: Local Secret "VAULT_TOKEN" was not found.
Is this the expected behavior? should all secrets be registered wherever flows are created/registered?Zanie
03/16/2021, 3:04 PMCharles Leung
03/16/2021, 3:06 PMfrom prefect.client import Secret
Secret("VAULT_TOKEN").get()
Using the old secret methodZanie
03/16/2021, 3:06 PMget()
on the secret manually, that's where it tries to get the valueCharles Leung
03/16/2021, 3:07 PMZanie
03/16/2021, 3:08 PMflow.storage = Github(..., secrets=["VAULT_TOKEN"])
PrefectSecret("VAULT_TOKEN").get()
I believeCharles Leung
03/16/2021, 3:09 PMZanie
03/16/2021, 3:13 PMconfig.toml
during development)Charles Leung
03/16/2021, 3:15 PMZanie
03/16/2021, 3:17 PMSecret.get()
code will be called and if it errors you won't reach registration?stored_as_script
to storage and the entire script is executed rather than being pickledCharles Leung
03/16/2021, 4:01 PMZanie
03/16/2021, 5:01 PMCharles Leung
03/16/2021, 5:04 PM# Get Job Credentials from Vault
vault = hvac.Client(url='<URL>', verify=f'<cert>')
aws_config = vault.read('<path>').get('data')
smb_config = vault.read('<path>').get('data')
smb_config.update(dict(
my_name=socket.getfqdn(),
use_ntlm_v2=True
))
@prefect.task
def upload_smb_files():
# Get File Difference from SMB to S3
# Setup
s3_client = boto3.client('s3', **aws_config)
smb_connection = SMBConnection(
remote_name='<remote>',
**smb_config
)
smb_connection.connect('')
try:
# move files with config
finally:
smb_connection.close()
return len(files)
flow.run_config = DockerRun(
image='<image>'
)
flow.storage = GitLab(
host="<gitlab host>",
repo="<repo path>", # name of repo
path="<path in repo>", # location of flow file in repo
access_token_secret="GITLAB_ACCESS_TOKEN", # name of personal access token secret
)
Zanie
03/16/2021, 5:20 PMfrom prefect import Flow, task
x = "FOO"
@task(log_stdout=True)
def display(value):
print(value)
with Flow("constant-global") as flow:
display(x)
serialized_flow = flow.serialize()
from pprint import pprint
pprint(serialized_flow)
❯ python example-constant-global.py
OrderedDict([('name', 'constant-global'),
('type', 'prefect.core.flow.Flow'),
('schedule', None),
('parameters', []),
('tasks',
[{'__version__': '0.14.11+10.gb0a47a530.dirty',
'auto_generated': False,
'cache_for': None,
'cache_key': None,
'cache_validator': {'fn': 'prefect.engine.cache_validators.never_use',
'kwargs': {}},
'inputs': {'value': {'required': True, 'type': 'typing.Any'}},
'max_retries': 0,
'name': 'display',
'outputs': 'typing.Any',
'retry_delay': None,
'skip_on_upstream_skip': True,
'slug': 'display-1',
'tags': [],
'timeout': None,
'trigger': {'fn': 'prefect.triggers.all_successful',
'kwargs': {}},
'type': 'prefect.tasks.core.function.FunctionTask'}]),
('edges', []),
('reference_tasks', []),
('environment', None),
('run_config', None),
('__version__', '0.14.11+10.gb0a47a530.dirty'),
('storage', None)])
from prefect import Flow, task, Parameter
x = "FOO"
@task(log_stdout=True)
def display(value):
print(value)
with Flow("constant-global") as flow:
x_param = Parameter("x", default=x)
display(x_param)
serialized_flow = flow.serialize()
from pprint import pprint
pprint(serialized_flow)
would store your value in the Prefect API
('parameters',
[{'__version__': '0.14.11+10.gb0a47a530.dirty',
'default': 'FOO',
'name': 'x',
'outputs': 'typing.Any',
'required': False,
'slug': 'x',
'tags': [],
'type': 'prefect.core.parameter.Parameter'}]),
Charles Leung
03/16/2021, 5:24 PM