https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I try to made run_deployment as an ac...
# ask-marvin
y
@Marvin I try to made run_deployment as an action, the flow accepts the triggered flow runtime data.
Copy code
def _run_deployment_action(flow_name: str, deployment_name: str):
    with get_client(sync_client=True) as client:
        deployment = client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
    deployment_id = str(deployment.id)
    return RunDeployment(
        source="selected",
        deployment_id=deployment_id,
        parameters={
            "flow_run_id": "{{ flow_run.id }}",
            "flow_run_name": "{{ flow_run.name }}",
            "flow_run_state": "{{ flow_run.state.name }}",
        },
        job_variables={},
    )
but it is showing `
Copy code
"reason": "Unable to create flow run from deployment: InvalidJinja()",
`
m
✅ thought for 123.9 seconds
{{ flow_run.id }}
can be valid in a 
RunDeployment
action, but the error suggests the automation can't successfully render one of your Jinja expressions in that action context. A few important points: In automations, Jinja context depends on the triggering event. If the event does not include a 
flow_run
resource, then 
flow_run.*
is not available. Your example:
Copy code
def _run_deployment_action(flow_name: str, deployment_name: str):
    with get_client(sync_client=True) as client:
        deployment = client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
    deployment_id = str(deployment.id)
    return RunDeployment(
        source="selected",
        deployment_id=deployment_id,
        parameters={
            "flow_run_id": "{{ flow_run.id }}",
            "flow_run_name": "{{ flow_run.name }}",
            "flow_run_state": "{{ flow_run.state.name }}",
        },
        job_variables={},
    )
The most likely issue is: * 
flow_run.state.name
is not always safe/available in the action template context for your triggering event * or your automation is triggered by an event that does not expose 
flow_run
I'd recommend first simplifying the parameters to confirm which field breaks rendering:
Copy code
return RunDeployment(
    source="selected",
    deployment_id=deployment_id,
    parameters={
        "flow_run_id": "{{ flow_run.id }}",
        "flow_run_name": "{{ flow_run.name }}",
    },
    job_variables={},
)
If that works, then 
flow_run.state.name
is the culprit. You can also try using event data directly, depending on your trigger. For example:
Copy code
parameters={
    "event_type": "{{ event.event }}",
    "occurred": "{{ event.occurred }}",
    "flow_run_id": "{{ flow_run.id }}",
}
A few debugging tips: 1. Check what event triggers the automation If it's not a 
prefect.flow-run.*
event, 
flow_run
may not exist. 2. Remove fields one by one Start with only:
Copy code
{
  "flow_run_id": "{{ flow_run.id }}"
}
3. Prefer simple fields first Nested state access like 
flow_run.state.name
may be more fragile than 
flow_run.id
/ 
flow_run.name
. If you want, paste the automation trigger/event type you're using, and I can tell you exactly which template fields are valid there. Relevant docs: * Automations docs * Automation guides
y
@Marvin this is the trigger
Copy code
def create_p1_automation() -> str:
    """P1: failure or >30min running → run on-call flow + critical alert."""
    logger = get_logger(__name__)

    automation = Automation(
        name="singularity_p1_critical",
        description=("Tagged singularity_p1: on Failed/Crashed or Running >30min, "
                     "run the on-call flow and post to singularity-critical-failures."),
        enabled=True,
        trigger=CompoundTrigger(
            require="any",
            triggers=[
                build_tag_based_failure_trigger(['singularity_p1']),
                build_tag_based_long_running_trigger(['singularity_p1']),
            ],
        ),
        actions=[
            _run_deployment_action('singularity_on_call_flow', 'singularity_on_call_flow'),
            _send_notification_action(
                P1_BLOCK_NAME,
                subject="[P1] Singularity flow failure",
            ),
        ],
    )
    created_id = automation.create()
    <http://logger.info|logger.info>("Created P1 automation id=%s", created_id)
    return str(created_id)
m
✅ thought for 114.3 seconds
Thanks — that helps. The likely problem is your long-running trigger. With a 
CompoundTrigger(require="any", ...)
, your action can be fired by either: 1. the failure trigger 2. the long-running/proactive trigger For the failure case, 
flow_run
is usually available, so things like:
Copy code
{{ flow_run.id }}
often work. For the long-running/proactive case, the automation may not have a normal 
triggering_event
carrying full 
flow_run
state context. That means expressions like:
Copy code
{{ flow_run.id }}  
{{ flow_run.name }}  
{{ flow_run.state.name }}
can fail to render for that branch, which matches the 
InvalidJinja()
error you're seeing. So the issue is probably not the compound trigger itself, but that one of its child triggers does not provide 
flow_run
in the Jinja context. What to do instead: *For proactive / long-running automations, prefer 
labels
or 
event
fields over 
flow_run.*
when possible.* A safer pattern is to pass the resource id from labels:
Copy code
parameters={
    "source_resource_id": "{{ labels['prefect.resource.id'] }}",
    "event_type": "{{ event.event if event else '' }}",
}
If you specifically want the flow run id, and your label contains something like 
prefect.flow-run.<uuid>
, you can extract it:
Copy code
parameters={
    "flow_run_resource_id": "{{ labels['prefect.resource.id'] }}",
}
Then parse that in the called flow. If you want to keep your current approach, the simplest test is:
Copy code
return RunDeployment(
    source="selected",
    deployment_id=deployment_id,
    parameters={
        "flow_run_id": "{{ labels['prefect.resource.id'] }}",
    },
    job_variables={},
)
If that works, you've confirmed the issue is with 
flow_run.*
availability in the long-running branch. You can also split this into two separate automations: * one for failed/crashed flow runs using 
flow_run.*
* one for long-running flows using 
labels[...]
That is often the cleanest solution because the template context is different for reactive vs proactive triggers. So my recommendation: Best option: split the automation in two. If you want one shared action, make it safe for both branches by using only context guaranteed in both, especially 
labels
. Something like:
Copy code
def _run_deployment_action(flow_name: str, deployment_name: str):
    with get_client(sync_client=True) as client:
        deployment = client.read_deployment_by_name(f"{flow_name}/{deployment_name}")
    deployment_id = str(deployment.id)
    return RunDeployment(
        source="selected",
        deployment_id=deployment_id,
        parameters={
            "source_resource_id": "{{ labels['prefect.resource.id'] }}",
            "event_type": "{{ event.event if event else '' }}",
        },
        job_variables={},
    )
Then the downstream flow can derive the run id from 
source_resource_id
. If you want, paste the implementations of: * 
build_tag_based_failure_trigger
* 
build_tag_based_long_running_trigger
and I can tell you exactly which fields are safe to use for each one.
3 Views
Open in Slack
PreviousNext
Powered by