<@ULVA73B9P> Can we link separate flows in prefect...
# ask-marvin
i
@Marvin Can we link separate flows in prefect? Meaning that once one flow is done, it triggers the next one to run
m
It is indeed possible to link separate flows in Prefect, so that once one flow completes, it triggers the next flow to run. The method for doing so depends on the version of Prefect you are using. Below are the steps based on the version: ### Prefect 2.x In Prefect 2.x, you can use deployment and automation features to create compound triggers that can initiate a flow based on the completion of other flows. Here is an example that shows how to set up this kind of dependency:
Copy code
python
{
 "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": "daily-customer-export",
       "prefect.resource.role": "flow"
     }
   },
   {
     "type": "event",
     "posture": "Reactive",
     "expect": ["prefect.block.remote-file-system.write_path.called"],
     "match_related": {
       "prefect.resource.name": "daily-revenue-export",
       "prefect.resource.role": "flow"
     }
   },
   {
     "type": "event",
     "posture": "Reactive",
     "expect": ["prefect.block.remote-file-system.write_path.called"],
     "match_related": {
       "prefect.resource.name": "daily-expenses-export",
       "prefect.resource.role": "flow"
     }
   }
 ]
}
Source: Prefect 2.x Automations Documentation ### Prefect 3.x In Prefect 3.x, you can use the
flow.from_source
method to orchestrate multiple flows. Additionally, you can create custom triggers to initiate flows based on events from other flows:
Copy code
python
from prefect import deploy, flow

# Define local flow
@flow(log_prints=True)
def local_flow():
    print("I'm a flow!")

# Deploy flows
if __name__ == "__main__":
    deploy(
        local_flow.to_deployment(name="example-deploy-local-flow"),
        flow.from_source(
            source="<https://github.com/org/repo.git>",
            entrypoint="flows.py:my_flow",
        ).to_deployment(
            name="example-deploy-remote-flow",
        ),
        work_pool_name="my-docker-pool",
        image="my-registry/my-image:dev",
    )
For custom triggers, you could define event-based triggers to automate flow dependencies:
Copy code
text
{
 "expect": ["prefect.block.remote-file-system.write_path.called"],
 "match_related": {
   "prefect.resource.name": "daily-customer-export",
   "prefect.resource.role": "flow"
 }
},
# ... (additional triggers for other flows)
Source: Prefect 3.x Custom Triggers Documentation To implement these solutions, ensuring your Prefect configuration and deployment setup is correct is crucial. If you have any more specific questions, feel free to ask!
n
this examples a bit more complicated than it needs to be, you can use run_deployment or a simple event trigger as shown here
👍 1