<@ULVA73B9P> can I pass the deployment context to ...
# marvin-ai
f
@Marvin can I pass the deployment context to a flow hook method executed on_completion? The doc only mentions flow, _flow_run_, state being passed
m
thought for 297.0 seconds
Short answer: you can’t change the hook signature—Prefect will only pass
flow
,
flow_run
, and
state
. But you can still access deployment context from inside the hook. Ways to get deployment info in an on_completion hook (Prefect 3.x): - From the runtime context: -
prefect.runtime.deployment
exposes
id
,
name
,
version
,
parameters
(None/empty if not running from a deployment) -
prefect.runtime.flow_run
exposes run fields like
id
,
name
,
parameters
, etc. - Docs: Runtime context, runtime.deployment API, runtime.flow_run API - From the
flow_run
argument: -
flow_run.deployment_id
,
flow_run.deployment_version
, plus other metadata fields - If you need to pass extra, custom data to the hook: - Pre-bind it when registering the hook (via
functools.partial
) - Or stash it during the flow run and read it in the hook (same process only) References for hooks: State-change hooks, Flow decorator and hooks API Examples - Access deployment context directly in the hook
Copy code
from prefect import flow
from prefect.runtime import deployment

def on_complete(flow, flow_run, state):
    # Via runtime context (works only if triggered from a deployment)
    print("Deployment name:", deployment.name)
    print("Deployment id:", deployment.id)
    # Via flow_run object
    print("Deployment id from flow_run:", flow_run.deployment_id)
    print("Deployment version from flow_run:", flow_run.deployment_version)

@flow(on_completion=[on_complete])
def my_flow():
    ...
- Pass extra configuration to the hook
Copy code
from functools import partial
from prefect import flow

def notifier(flow, flow_run, state, team="data-eng"):
    print(f"[{team}] Flow {flow.name} finished in state {state.name}")

@flow(on_completion=[partial(notifier, team="mlops")])
def my_flow():
    ...
- Stash data during the run for the hook to read later (same process)
Copy code
from prefect import flow
from prefect.context import get_run_context

def on_complete(flow, flow_run, state):
    cfg = flow_run.context.get("hook_cfg", {})
    print("Hook cfg:", cfg)

@flow(on_completion=[on_complete])
def my_flow():
    ctx = get_run_context()
    ctx.flow_run.context["hook_cfg"] = {"alert_level": "critical"}
    ...
If you can share which “deployment context” fields you need (e.g., name, id, parameters, work pool/queue), I can suggest the most direct access pattern.
f
@Marvin only the deployment name
m
thought for 27.9 seconds
Yes—while the hook signature can’t be changed (it only receives
flow
,
flow_run
,
state
), you can get the deployment name inside the hook. Fast path (when running from a deployment):
Copy code
from prefect.runtime import deployment

def on_complete(flow, flow_run, state):
    dep_name = deployment.name  # None if not running via a deployment
    print("Deployment name:", dep_name)
Robust path (fallback to the API if needed):
Copy code
from prefect.runtime import deployment

async def on_complete(flow, flow_run, state):
    name = deployment.name
    if not name and flow_run.deployment_id:
        from prefect.client.orchestration import get_client
        async with get_client() as client:
            dep = await client.read_deployment(flow_run.deployment_id)
            name = dep.name
    print("Deployment name:", name)
Docs: - State-change hooks - Runtime context - runtime.deployment API