<@ULVA73B9P> Any way to pass custom/runtime parame...
# marvin-ai
j
@Marvin Any way to pass custom/runtime parameters to on_failure decorator hook? Without attaching the with_options to the flow in advance?
m
thought for 200.6 seconds
Short answer: not directly. In Prefect 3.x, hook functions have a fixed signature and are invoked with three arguments:
(flow, flow_run, state)
. You can’t pass arbitrary, ad-hoc args into a hook at call time via the decorator. The usual way to “parameterize” a hook is either: - Bind kwargs with
functools.partial
when attaching the hook (e.g., via
with_options
), or - Read whatever you need from runtime objects available to the hook. Good news: you can get most runtime data inside the hook without `with_options`: - `flow_run.parameters`: all parameters passed to the flow run - `flow_run.tags`: tags on the run - `flow_run.name`: run name - `get_run_context()`: full runtime context; includes
ctx.parameters
,
ctx.flow_run
,
ctx.flow
, and a client you can use to fetch Variables, etc. Examples - Attach via decorator and read runtime data in the hook:
Copy code
from prefect import flow
from prefect.context import get_run_context

def notify_on_failure(flow, flow_run, state):
    # Access runtime data without passing custom args
    params = flow_run.parameters        # or get_run_context().parameters
    tags = flow_run.tags
    run_name = flow_run.name
    # do something with params/tags/run_name/state.message ...

@flow(on_failure=[notify_on_failure])
def my_flow(user_id: str, priority: str = "normal"):
    raise RuntimeError("boom")

my_flow(user_id="abc123", priority="high")
- If you need truly custom per-run data that isn’t a flow param, you can: - Put it in a tag and read
flow_run.tags
- Set it as part of the run’s context when creating the run via API (if you’re starting runs programmatically), then read
get_run_context().flow_run.context
- Store it in a Prefect Variable or env var and read it inside the hook What you cannot do - You can’t pass extra positional/keyword args to a hook attached via
@flow(..., on_failure=[...])
or
@my_flow.on_failure
at run time. - You can’t register hooks dynamically after execution starts; they must be attached before the state change occurs. Docs - State change hooks guide: https://docs.prefect.io/latest/how-to/workflows/state-change-hooks/ - Runtime context (
get_run_context
): https://docs.prefect.io/latest/api-ref/context/#prefect.context.get_run_context If you can share what custom data you’re trying to pass, I can suggest the cleanest pattern (e.g., parameters vs tags vs variables vs context) for your setup.
j
@Marvin Can I set flow run context during runtime and pick/retrieve it later in the hook?
m
thought for 160.5 seconds
Short answer: not during the run. In Prefect 3.x, the flow run context you see in hooks is read-only. You can read from it in your hook, but you cannot mutate it mid-run and then pick it up later in the same run. What you can do instead: - Set context at creation time: If you start the run programmatically, you can provide a context dict when creating the flow run. Your hook can read it via
flow_run.context
. - Use flow parameters: Pass any per-run data as parameters; your hook can read them from
flow_run.parameters
or
get_run_context().parameters
. - Use job variables: Deployment-level values are available via runtime and can be read in hooks. - Persist mutable data externally: If you need mutable, ad-hoc data that changes during the run and is later read by a hook, write it to an external store keyed by
flow_run.id
(e.g., DB/Redis/object storage) and read it in the hook. Example hook accessing runtime data without custom args:
Copy code
from prefect import flow
from prefect.context import get_run_context

def on_fail(flow, flow_run, state):
    ctx = get_run_context()
    # Read-only views of runtime data
    params = ctx.parameters            # same as flow_run.parameters
    tags = flow_run.tags
    creation_context = flow_run.context  # set at flow-run creation time
    # fetch mutable extras from external store if needed using flow_run.id
    # ...

@flow(on_failure=[on_fail])
def my_flow(user_id: str, feature_flag: str):
    raise RuntimeError("boom")
If you’re starting runs via the API and want to seed
flow_run.context
, you can pass a dict at creation time; your hook can read it later. But there’s no supported API to update
flow_run.context
mid-run. If you tell me what data you want to stash and when you need it, I can suggest the cleanest pattern (params vs context-at-creation vs external store).