Zhibin Dai
03/03/2022, 5:08 PMrun_env = Parameter("run_env", default="DEV")
sf_user = PrefectSecret(SF_USER + "_" + run_env)
Kevin Kho
03/03/2022, 5:09 PMSF_USER + "_" + run_env
was a task output because that is evaluated during build time instead of runtimeZhibin Dai
03/03/2022, 5:25 PM@task
def concat_str(str1, str2, delim):
return str1 + delim + str2
with Flow(FLOW_NAME, storage=STORAGE, run_config=RUN_CONFIG,) as flow:
run_env = Parameter(PREFECT_RUN_ENVIRONMENT_PARAM_NAME, default='DEV')
sf_user = PrefectSecret(concat_str(SF_USERNAME, run_env, "_"))
And it get this error
Traceback (most recent call last):
File "/opt/homebrew/lib/python3.9/site-packages/prefect/cli/build_register.py", line 134, in load_flows_from_script
namespace = runpy.run_path(abs_path, run_name="<flow>")
File "/opt/homebrew/Cellar/python@3.9/3.9.9/Frameworks/Python.framework/Versions/3.9/lib/python3.9/runpy.py", line 268, in run_path
return _run_module_code(code, init_globals, run_name,
File "/opt/homebrew/Cellar/python@3.9/3.9.9/Frameworks/Python.framework/Versions/3.9/lib/python3.9/runpy.py", line 97, in _run_module_code
_run_code(code, mod_globals, init_globals,
File "/opt/homebrew/Cellar/python@3.9/3.9.9/Frameworks/Python.framework/Versions/3.9/lib/python3.9/runpy.py", line 87, in _run_code
exec(code, run_globals)
File "/Users/zhibin/Desktop/repos/parcl-orchestration/flows/ppi/s3_to_sf_raw.py", line 50, in <module>
sf_user = PrefectSecret(concat_str(SNOWFLAKE_PREFECT_USERNAME, run_env, "_"))
File "/opt/homebrew/lib/python3.9/site-packages/prefect/core/task.py", line 159, in init
old_init(self, *args, **kwargs)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/tasks/secrets/base.py", line 45, in __init__
super().__init__(name=name, **kwargs)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/core/task.py", line 159, in init
old_init(self, *args, **kwargs)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/tasks/secrets/base.py", line 26, in __init__
super().__init__(**kwargs)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/core/task.py", line 159, in init
old_init(self, *args, **kwargs)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/core/task.py", line 345, in __init__
self.logger = logging.get_logger(self.name)
File "/opt/homebrew/lib/python3.9/site-packages/prefect/utilities/logging.py", line 309, in get_logger
return prefect_logger.getChild(name)
File "/opt/homebrew/Cellar/python@3.9/3.9.9/Frameworks/Python.framework/Versions/3.9/lib/python3.9/logging/__init__.py", line 1727, in getChild
suffix = '.'.join((self.name, suffix))
TypeError: sequence item 1: expected str instance, FunctionTask found
Kevin Kho
03/03/2022, 5:47 PMsf_user = PrefectSecret()(concat_str(SF_USERNAME, run_env, "_"))
really not sure it will work\Zhibin Dai
03/03/2022, 5:52 PM└── 12:50:50 | INFO | Entered state <Failed>: Failed to load and execute flow run: TypeError('sequence item 1: expected str instance, FunctionTask found')
└── 12:50:51 | ERROR | Failed to load and execute flow run: TypeError('sequence item 1: expected str instance, FunctionTask found')
Flow run failed!
sf_user = PrefectSecret()(concat_str(SNOWFLAKE_PREFECT_USERNAME, run_env, "_"))
<http://logger.info|logger.info>(f"SF USER: {sf_user}")
Kevin Kho
03/03/2022, 5:55 PMfrom prefect import Flow, task
import prefect
from prefect.tasks.secrets import PrefectSecret
@task
def t():
return "SLACK_WEBHOOK_URL"
@task
def log(x):
<http://prefect.context.logger.info|prefect.context.logger.info>(x)
return 1
with Flow("...") as flow:
x = t()
a = PrefectSecret()(x)
log(a)
flow.run()