David DeStefano
03/11/2025, 8:34 PMMarvin
03/11/2025, 8:34 PMDeploymentTrigger
. You can match events from an upstream flow to trigger the downstream flow.
2. Passing Parameters: When you trigger a flow via the Prefect API, you must pass parameters as keyword arguments. Ensure that these parameters are validated and do not exceed 512kb in size.
3. Example Setup:
- Define your upstream and downstream flows using the @flow
decorator.
- Use tags
to identify the upstream flow runs you want to trigger the downstream flow from.
- Configure DeploymentTrigger
in your downstream flow to listen for specific events from the upstream flow.
Here's a simplified example based on the gathered information:
python
from prefect import flow, tags
from prefect.events.schemas import DeploymentTrigger
TAG_NAME = "example_tag"
@flow
def upstream_flow(param1):
# Your upstream logic
pass
@flow
def downstream_flow(param2):
# Your downstream logic
pass
if __name__ == "__main__":
with tags(TAG_NAME):
upstream_flow(param1="value1")
downstream_flow.serve(
name="downstream",
triggers=[
DeploymentTrigger(
expect={"prefect.flow-run.Completed"},
match_related=[
{
"prefect.resource.role": "tag",
"prefect.resource.id": f"prefect.tag.{TAG_NAME}",
},
{
"prefect.resource.role": "flow",
"prefect.resource.name": "upstream_flow",
}
]
)
]
)
For more details, you can refer to the Prefect documentation on writing flows and automations and triggers.
If you need further assistance or examples, feel free to ask.