https://prefect.io logo
Title
y

YZ

10/06/2022, 5:38 PM
Hi folks, for a
state_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
r

Rob Freedy

10/06/2022, 6:25 PM
Hey @YZ! A good example of this can be found in this post for 1.0. I believe you can access the parameters for the task through the context like this:
prefect.context["parameters"].*get*("parameter_name")
y

YZ

10/06/2022, 6:32 PM
Interesting, here
is_prod
is not a Parameter
r

Rob Freedy

10/06/2022, 6:42 PM
So what is in config that you are trying to access? State handlers only pass in the Flow/Task object and the old and new states: https://docs-v1.prefect.io/core/concepts/notifications.html#state-handlers
y

YZ

10/06/2022, 6:43 PM
config is a dictionary loaded from a yaml file and combined with some run time env variables.
Do you have an example of the task/flow object being used inside the state_handlers? I only saw examples of old/new states being used
I can pass
config
to
task
, but I would like to use it inside the state_handler method.
I modified my pseu-code in my original question - hope it adds more clarity
r

Rob Freedy

10/06/2022, 8:35 PM
Are you able to make the config variable a Parameter and then access it through the context values like this example below? Here is some documentation on contexts: https://docs-v1.prefect.io/api/latest/utilities/context.html#context and https://docs-v1.prefect.io/core/concepts/execution.html#modifying-context-at-runtime
import 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)