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

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")

Interesting, here

is_prod

is not a Parameter

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

config is a dictionary loaded from a yaml file and combined with some run time env variables.

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)