YZ
10/06/2022, 5:38 PMstate_handler(task, old_state, new_state)
Is there a way to access the original input parameters for the task
? For example, my original task is as below, and I would like to access config
in the state_handler
def state_handler(task, old_state, new_state):
# Question: how can I access the `config` variable I originally passed into `task`?
@task(state_handlers=[state_handler])
def my_task(config: Any):
if config.is_prod:
# do something
Rob Freedy
10/06/2022, 6:25 PMprefect.context["parameters"].*get*("parameter_name")
YZ
10/06/2022, 6:32 PMis_prod
is not a ParameterRob Freedy
10/06/2022, 6:42 PMYZ
10/06/2022, 6:43 PMconfig
to task
, but I would like to use it inside the state_handler method.Rob Freedy
10/06/2022, 8:35 PMimport prefect
from prefect import Flow, task, Parameter
from prefect.tasks.prefect import RenameFlowRun
def rename_handler(obj, new_state, old_state):
if new_state.is_running():
user = prefect.context.parameters.get("user_name")
param_value = user.lower().replace(" ", "_")
RenameFlowRun().run(flow_run_name=f"hello_{param_value}")
return
@task(log_stdout=True)
def greet_user(user_name):
print(f"Hello {user_name}")
with Flow("using_parameters_in_state_handler", state_handlers=[rename_handler]) as flow:
param = Parameter("user_name", default="Jerry Seinfeld")
greet_user(param)