Is there a way to pass the value of a Parameter into a Prefe Prefect Community #prefect-server

Is there a way to pass the value of a Parameter in...

Zhibin Dai

03/03/2022, 5:08 PM

Is there a way to pass the value of a Parameter into a PrefectSecret as a string? Is something like this possible?

Copy code

run_env = Parameter("run_env", default="DEV")
sf_user = PrefectSecret(SF_USER + "_" + run_env)

Kevin Kho

03/03/2022, 5:09 PM

I believe yes if

SF_USER + "_" + run_env

was a task output because that is evaluated during build time instead of runtime

Zhibin Dai

03/03/2022, 5:25 PM

cool ill give it a shot

Zhibin Dai

03/03/2022, 5:44 PM

I tried this but it doesnt seem to work. Heres the code

Copy code

@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

Copy code

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 PM

can you try:

Copy code

sf_user = PrefectSecret()(concat_str(SF_USERNAME, run_env, "_"))

really not sure it will work\

Kevin Kho

03/03/2022, 5:49 PM

Actually I just read the source and I think it will work

Zhibin Dai

03/03/2022, 5:52 PM

Copy code

└── 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!

Zhibin Dai

03/03/2022, 5:53 PM

Copy code

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 PM

The logger won’t work in the flow, it needs to be in a task haha. One sec let me test

Kevin Kho

03/03/2022, 5:58 PM

Copy code

from 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()

Kevin Kho

03/03/2022, 5:58 PM

this works for me

2 Views

Open in Slack

Previous Next