I have been deploying my flows via something that ...
# ask-community
b
I have been deploying my flows via something that looks similar to below:
Copy code
from prefect import flow

if __name__ == "__main__":
    flow.from_source(
        source="my-github-repo.git",
        entrypoint="prefect/flows/etl.py:etl_flow",
    ).deploy(
        name="my-name",
        work_pool_name="my-wrork-pool",
        job_variables={"env": {"env1": "loves-to-code"},
                       "pip_packages": ["pandas", "requests"]},
        cron="15 0 * * *",
        tags=["prod"],
        description="my awesome flow description",
        version="1.0.0",
    )
I am writing a new flow (called flow2), but instead of a schedule, I want the flow to run after another flow (flow1) completes successfully. For each successful run for flow1, I want flow2 to to kickoff. It's not obvious to me how I can deploy similar to above and specify the dependent relationship.
n
hi @Brock - there are two main ways you can do this • my recommendation: put a trigger on the downstream one that will
expect
prefect.flow-run.Completed
events that
match_related
resources like the upstream's deployment name • otherwise you can always call
run_deployment
in a
on_completion
hook with
timeout=0
to avoid blocking until that downstream flow finishes (without timeout=0 you create an infinite blocking waterfall)
b
Thanks Nate, I really appreciate the note on a Sunday. In my case, because I have been using .deploy, would I add something similar to below in my deploy script's logic? In my case, I have already deployed the upstream flow and its been running on a schedule for a few weeks now.
Copy code
triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.Completed"},
                match_related={"prefect.resource.name": "upstream_deployment"},
            )
        ],
n
yep exactly right! you'd just replace this
match_related={"prefect.resource.name": "upstream_deployment"},
according to the actual name of your upstream deployment, or whatever feature of that deployment would be most stable to match on (bespoke tag, for example)
both
.serve
and
.deploy
accept triggers the same way
b
Fantastic thank you!
catjam 1
@Nate Just wanted to follow up quickly and say thanks again. I got it deployed and tested manually on the console. One question. Should I see the deployment name I specified flagged in the console? It shows "All Resources". I tested the deployment/flow and it successfully triggered this deployment. My deploy script was:
Copy code
).deploy(
        name="mlops-train-model",
        work_pool_name="brock-pool1",
        job_variables={"env": {"BROCK": "loves-to-code"},
                       "pip_packages": ["pandas", "requests"]},
        tags=["prod"],
        description="Pipeline to train a model and log metrics and parameters for a training job",
        version="1.0.0",
        triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.Completed"},
                match_related={"prefect.resource.name": "ml-datasets"}
            )
        ]
    )
where ml-datasets is the name of the deployment. I tested to ensure it wasn't triggered for an unrelated deployment/flow, and it behaved as expected. TLDR: From the interface, is there a way to visually confirm the logic?