Nicholas Pini
11/06/2025, 9:30 PMMarvin
11/06/2025, 9:30 PMMarvin
11/06/2025, 9:34 PMprefect.flow-run.Completed) and on resource/related-resource labels (flow, deployment, tags, etc.), but there isn’t a filter for state.name or state.message.
Good ways to achieve what you want:
- Emit a custom event when your “special completed” condition occurs, then trigger the automation on that event.
- Alternatively, conditionally add a tag and trigger on the Completed event only when that tag is present.
Option A: Custom event (recommended)
1) In your flow, detect the condition and emit a custom event before returning your custom Completed state.
from prefect import flow, get_run_context
from prefect.events import emit_event
from prefect.states import Completed
@flow
def my_flow():
# ... your logic
partial = True # example condition
if partial:
rc = get_run_context()
emit_event(
event="myapp.flow.partial_success",
resource={"prefect.resource.id": f"prefect.flow-run.{rc.flow_run.id}"},
related=[{"prefect.resource.id": f"prefect.flow.{rc.flow.name}"}],
payload={"details": "3 of 5 tasks succeeded"},
)
return Completed(name="Partial Success", message="3 of 5 tasks succeeded")
return Completed()
2) Create an automation with an Event trigger that expects myapp.flow.partial_success and add whatever action(s) you need (e.g., Send Notification).
Docs:
- Automations & event triggers
- States (custom names/messages)
- Events & webhooks (concepts)
Option B: Conditional tag + Completed event
- Only when your partial-success condition is true, add a tag and then return. Your Completed event will include the tag as a related resource, and your automation can “Expect: prefect.flow-run.Completed” and “Match related: that tag”.
- Example sketch:
from prefect import flow, tags
from prefect.states import Completed
@flow
def my_flow():
# ... logic
if partial_success:
with tags("partial-success"):
return Completed(name="Partial Success", message="3 of 5 tasks succeeded")
return Completed()
Caveats:
- The trigger system does not currently expose filters for a state’s custom name/message, so you can’t target “Completed(name='Partial Success')” directly.
- Custom events are the most reliable, explicit way to drive automations for nuanced outcomes.
If you’d like, tell me whether you prefer setting this up via the UI, CLI, or Python API and I can share step-by-step instructions for creating the automation.Nate
11/06/2025, 9:55 PMfrom prefect import flow, serve
from prefect.events import DeploymentEventTrigger
from prefect.states import Completed
@flow(log_prints=True)
def upstream_flow(should_trigger: bool = True):
print(f"Upstream flow running with should_trigger={should_trigger}")
if should_trigger:
return Completed(
message="This is a custom completion that should trigger downstream",
name="CustomCompleted"
)
else:
return Completed(
message="This is a regular completion that should NOT trigger downstream"
)
@flow(log_prints=True)
def downstream_flow():
print("Downstream flow triggered by custom state!")
if __name__ == "__main__":
upstream_deployment = upstream_flow.to_deployment(name="upstream")
downstream_deployment = downstream_flow.to_deployment(
name="downstream",
triggers=[
DeploymentEventTrigger(
expect={"prefect.flow-run.CustomCompleted"},
match={"prefect.resource.id": "prefect.flow-run.*"},
match_related={"prefect.resource.name": "upstream"},
)
],
)
serve(upstream_deployment, downstream_deployment)Marvin
11/06/2025, 9:56 PMMarvin
11/06/2025, 10:00 PMCompleted(name="CustomCompleted") emits the event prefect.flow-run.Completed, not prefect.flow-run.CustomCompleted.
- As a result, DeploymentEventTrigger(expect={"prefect.flow-run.CustomCompleted"}) will not fire — there is no such event name to match.
Docs to confirm:
- States: State types vs names
- Event triggers: Automations & event triggers
If your goal is to only trigger downstream on your “special” completion, here are two reliable patterns:
Option A: Emit a custom event explicitly (recommended)
from prefect import flow, serve, get_run_context
from prefect.events import DeploymentEventTrigger, emit_event
from prefect.states import Completed
@flow
def upstream_flow(should_trigger: bool = True):
if should_trigger:
rc = get_run_context()
emit_event(
event="myapp.flow.CustomCompleted", # your custom event
resource={"prefect.resource.id": f"prefect.flow-run.{rc.flow_run.id}"},
related=[{"prefect.resource.name": "upstream"}],
payload={"details": "This is a custom completion that should trigger downstream"},
)
return Completed(name="CustomCompleted")
else:
return Completed(message="Regular completion — no custom event")
@flow
def downstream_flow():
print("Downstream flow triggered by custom event!")
if __name__ == "__main__":
upstream_dep = upstream_flow.to_deployment(name="upstream")
downstream_dep = downstream_flow.to_deployment(
name="downstream",
triggers=[
DeploymentEventTrigger(
expect={"myapp.flow.CustomCompleted"},
match={"prefect.resource.id": "prefect.flow-run.*"},
match_related={"prefect.resource.name": "upstream"},
)
],
)
serve(upstream_dep, downstream_dep)
Option B: Conditional tag + Completed event
- Add a tag only when the special condition is true, and have your trigger “Expect: `prefect.flow-run.Completed`” but “Match related: tag=partial-success” to filter to those runs.
Given the above, I wouldn’t create a discussion saying “this is possible” because it would be incorrect and likely confuse others. However, I can create a discussion that:
- Clarifies that event names are based on state type (not custom name)
- Shows the working patterns (custom events or tags)
- Asks if folks want first-class support for filtering by custom state names/messages
Would you like me to open that discussion and include your snippet adapted to the custom-event approach?Nate
11/06/2025, 10:06 PMtriggering event
at 2025-11-06T22:02:35.309624Z, this event was emitted:
event: prefect.flow-run.CustomCompleted
flow_run: daring-mastiff (007ed9c9-3bf7-40f2-89da-b31b04636c71)
flow_name: upstream-flow
state: CustomCompleted
message: "This is a custom completion that should trigger downstream"
why it matched
1. resource pattern matched: the event came from a flow run resource
(prefect.flow-run.*)
2. event type matched: the event was prefect.flow-run.CustomCompleted (expected)
3. threshold met: 1 event was enough to trigger
4. related resource filter: the match_related filter looking for prefect.resource.name:
"upstream" matched against the flow name "upstream-flow"
action executed
the automation triggered a deployment run:
- deployment: downstream-flow/downstream
- resulting flow run: faithful-dog (completed successfully at 22:02:44)
so this is a classic upstream→downstream workflow orchestration pattern where a custom
completion state from the upstream flow automatically triggers the downstream flow via
automation.
lmk if you need to see more proof my friendMarvin
11/06/2025, 10:06 PMMarvin
11/06/2025, 10:08 PMNicholas Pini
11/07/2025, 8:21 AM