<@ULVA73B9P> how do I trigger a flow run based on ...
# ask-marvin
d
@Marvin how do I trigger a flow run based on the completion of a previous flow run?
m
To trigger a flow run based on the completion of a previous flow run, you can use Prefect Automations. Below is a step-by-step guide and some resources to help you achieve this: Step 1: Define an Automation Trigger Using the Automations feature, you can set up a trigger that initiates a flow run when a specific event (like the completion of a previous flow) occurs. Here’s an example of a compound trigger from the Prefect Automations documentation:
Copy code
json
{
  "type": "compound",
  "require": "all",
  "within": 3600,
  "triggers": [
    {
      "type": "event",
      "posture": "Reactive",
      "expect": ["prefect.block.remote-file-system.write_path.called"],
      "match_related": {
        "prefect.resource.name": "previous-flow-name",
        "prefect.resource.role": "flow"
      }
    }
  ]
}
Step 2: Use Dependencies in Your Flow For more complex scenarios, you may want to specify dependencies explicitly within your flows. This can be done by using
wait_for
in your flow and task definitions. Here’s a simple example:
Copy code
python
from prefect import flow, task, run_deployment

@task
def first_flow_task():
    # Some operations
    pass

@flow
def first_flow():
    first_flow_task()

@task
def second_flow_task():
    # Some operations
    pass

@flow
def second_flow():
    second_flow_task()

@flow
def parent_flow():
    first_run = run_deployment(name="first_flow")
    second_run = run_deployment(name="second_flow", wait_for=[first_run])

if __name__ == "__main__":
    parent_flow()
In this structure, the
second_flow
will only start after
first_flow
completes. Further reading can be found in the Prefect Upstream Dependencies guide. For additional context and examples, you may check the following relevant pull requests and issues: - Automations Guide PR: link - Tutorial section docs: link These resources should help you set up your desired workflow for triggering flows based on the completion of other flows.
n
+1 to trigger, but you dont need a compound one unless you have more complex logic https://github.com/PrefectHQ/prefect-recipes/tree/main/flows-advanced/using_state_hooks
dead simplest way is
run_deployment
in an
on_completion
hook with
timeout=0
to avoid blocking / waiting for that my preferred way is to configure a trigger on the downstream deployment that will
expect
a
prefect.flow-run.Completed
event from the upstream deployment, then its no code changes
d
Ok great thanks for that Nate will give it a go.
My problem is I need to trigger 3 different deployments for a day with 3 different sets of data and then once that day is complete I need to trigger the next day with the next 3 days of data if that makes sense? Would that be possible with this method? https://github.com/PrefectHQ/prefect-recipes/blob/main/flows-advanced/using_state_hooks/run_deployment_on_completion.py
n
I need to trigger 3 different deployments for a day with 3 different sets of data and then once that day is complete I need to trigger the next day with the next 3 days of data
hmm im not sure I totally understand. I would say it depends on where you're getting the data you want to pass to downstream deployments and whether or not you can write code to run in the hook that would retrieve it
d
The data is coming from a GCS bucket and I was thinking to create a python script that does this all so I would be able to write code in the hook
So it would work like • day 1: ◦ data1 -> flow-run1 ◦ data2 -> flow-run2 ◦ data3 -> flow-run3 • day2 - begin on completion of previous 3 flows: ◦ day2 data1 -> flow-run1 ◦ day2 data2 -> flow-run2 ◦ day2 data3 -> flow-run3 • etc. for the rest of the days