https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How could I deploy one flow A to be r...
# ask-marvin
l
@Marvin How could I deploy one flow A to be run every N mintues and another flow B to be run always after flow A is completed?
m
✅ thought for 156.4 seconds
Yes — in Prefect 3.x, the cleanest way is: * Deploy Flow A with a schedule (
interval
/ 
cron
) * Deploy Flow B with a trigger that listens for Flow A completion You generally have two good patterns: 1. Separate deployments + trigger/automation → best if A and B should stay independent 2. Call B from inside A as a subflow → best if B is always part of A and needs A's data directly --- Recommended: separate deployments This gives you: * Flow A runs every N minutes * Flow B starts only when A completes * A and B can be managed/retried/observed independently Example:
Copy code
python
from datetime import timedelta

from prefect import flow
from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger

@flow
def flow_a():
    print("running A")

@flow
def flow_b():
    print("running B")

if __name__ == "__main__":
    flow_a.deploy(
        name="flow-a-every-5-min",
        work_pool_name="my-work-pool",
        interval=timedelta(minutes=5),
    )

    flow_b.deploy(
        name="flow-b-after-a",
        work_pool_name="my-work-pool",
        triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.completed"},
                match={
                    # match Flow A's events
                    # use the appropriate resource identifier for A
                    "prefect.resource.name": "flow-a"
                },
            )
        ],
    )
A few notes: * 
Interval
/ 
Cron
scheduling is available in 
prefect.schedules
* deployment triggers are available in 
prefect.events.schemas.deployment_triggers
* 
prefect deploy
is the supported CLI in Prefect 3.x --- If B should only run after A succeeds Be careful with the event you match. 
prefect.flow-run.completed
means the run reached a terminal completed state. If you specifically need “successful completion”, that is usually the right event to use for “finished successfully”; if your use case depends on state nuances, I can help you write the exact trigger filter. --- Alternative: make B a subflow of A If B should always happen after A and you want to pass Python objects/results directly, simplest is:
Copy code
python
from prefect import flow

@flow
def flow_b(x):
    print(f"running B with {x}")

@flow
def flow_a():
    result = {"value": 42}
    flow_b(result)

if __name__ == "__main__":
    flow_a.deploy(
        name="flow-a-every-5-min",
        work_pool_name="my-work-pool",
        interval=300,
    )
This is better when: * B is tightly coupled to A * B needs A's in-memory result directly * you do not need separate deployment management --- Which pattern should you choose? Use separate deployments + trigger if: * A and B are logically separate workflows * you want separate observability/retries/deployment configs * you may want other things to trigger B later Use subflow calling if: * B is always part of A * you need to pass rich Python data directly * you want the simplest implementation --- Important caveat about passing data If A and B are separate deployments, B does not automatically receive A's return value. Options: * pass static trigger parameters to B * write A's output to external storage / a database / object store, then let B read it * use a subflow instead if direct passing is required --- CLI note If you prefer the CLI, use 
prefect deploy
in Prefect 3.x, not the old 2.x deployment build command. Docs / app: * Prefect docs * Prefect Cloud If you want, I can give you a complete working example for either: * Python-only deployments with 
flow.deploy()
, or * 
prefect.yaml
+ 
prefect deploy
l
@Marvin I use prefect.yaml
m
✅ thought for 6.6 seconds
Perfect — with 
prefect.yaml
, the usual setup is: * Deployment A has a schedule every 
N
minutes * Deployment B has an event trigger that fires when A completes A simplified example looks like this:
Copy code
yaml
deployments:
  - name: flow-a
    entrypoint: flows.py:flow_a
    work_pool:
      name: my-work-pool
    schedule:
      interval: 300

  - name: flow-b
    entrypoint: flows.py:flow_b
    work_pool:
      name: my-work-pool
    triggers:
      - enabled: true
        match:
          prefect.resource.name: flow-a
        expect:
          - prefect.flow-run.completed
And your `flows.py`:
Copy code
python
from prefect import flow

@flow
def flow_a():
    print("running A")

@flow
def flow_b():
    print("running B after A")
Then deploy with:
Copy code
bash
prefect deploy
A few important notes: * 
flow-a
gets scheduled every 300 seconds * 
flow-b
is not scheduled directly — it is triggered by A's completion event * if you want B to run only after a specific deployment of A, you should match more specifically than just the flow name * if B needs A's output, separate deployments will not automatically pass return values between them If you want, I can write you a full valid 
prefect.yaml
example for: 1. match by flow name, or 2. match by deployment name (usually safer).
l
@Marvin Would it be possible to have two different deployment with the same name but each one pointing to different flows?
m
✅ thought for 13.0 seconds
Generally, no — you should not do that. In Prefect, a deployment name is meant to identify a specific deployment for a specific flow. Reusing the same deployment name for two different flows will be confusing at best, and may not behave the way you want. A safer mental model is: * flow name identifies the flow * deployment name identifies how that specific flow is deployed So this is good:
Copy code
yaml
deployments:
  - name: every-5-min
    entrypoint: flows.py:flow_a

  - name: after-a
    entrypoint: flows.py:flow_b
Even though the deployment names differ, they can still be related clearly. What you can do is have the same deployment name reused across different flows in some contexts, but Prefect typically refers to deployments as something like: 
flow-name/deployment-name
So the full identity is effectively the combination of both. That said: * it is still a bad idea operationally * it makes CLI usage, UI searching, automations, and debugging harder * matching triggers by name becomes ambiguous Recommended: keep deployment names unique and descriptive, for example:
Copy code
yaml
deployments:
  - name: flow-a-every-5-min
    entrypoint: flows.py:flow_a

  - name: flow-b-after-flow-a
    entrypoint: flows.py:flow_b
If you want, I can also show you how to configure the trigger in 
prefect.yaml
so flow-b-after-flow-a runs only after flow-a-every-5-min completes.
✅ thought for 104.4 seconds
Yes — that is legal in Prefect 3.x. Your example is valid:
Copy code
yaml
deployments:
  - name: flow-prod
    entrypoint: flows.py:flow_a

  - name: flow-prod
    entrypoint: flows.py:flow_b
The reason is that deployment names are not globally unique. They only need to be unique per flow. In practice, Prefect identifies deployments as: 
flow_name/deployment_name
So if your flows are named 
flow_a
and 
flow_b
, these become effectively: * 
flow_a/flow-prod
* 
flow_b/flow-prod
So yes, Prefect accepts this. Practical caveat: it can be ambiguous when using the CLI unless you specify the full name. For example:
Copy code
bash
prefect deploy --name flow-prod
may match both deployments. Instead, use:
Copy code
bash
prefect deploy --name flow_a/flow-prod
prefect deploy --name flow_b/flow-prod
or deploy all from the file with:
Copy code
bash
prefect deploy
Bottom line: Legal? Yes. Recommended? Usually only if you intentionally want the same deployment label across multiple flows. Relevant docs: * prefect.yaml deployments guide * Deployments concept docs If you want, I can show you how to write the trigger in 
prefect.yaml
so 
flow_b/flow-prod
runs after 
flow_a/flow-prod
finishes.
l
@Marvin Is the following legal?
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a

  - name: prod
    entrypoint: flow_b.py:flow_b
Since the full deployment name is always preceded by the flow name as 
flow-name/
?
m
✅ thought for 6.5 seconds
Yes — that is legal in Prefect 3.x. Your two deployments are distinct because Prefect treats the effective identity as: * 
flow_a/prod
* 
flow_b/prod
not just 
prod
. So this is valid:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a

  - name: prod
    entrypoint: flow_b.py:flow_b
A couple of practical notes: * In the UI and API, deployments are commonly referenced as 
flow-name/deployment-name
* In the CLI, 
prefect deploy --name prod
may be ambiguous if both exist * To target one specifically, use the full name:
Copy code
bash
prefect deploy --name flow_a/prod
prefect deploy --name flow_b/prod
So your reasoning is correct: because the full name is effectively prefixed by the flow name, duplicate deployment 
name
values across different flows are allowed. Docs: * Deployments concepts * Using prefect.yaml If you'd like, I can also show you how to configure 
flow_b/prod
to trigger only after 
flow_a/prod
completes in 
prefect.yaml
.
l
@Marvin How could I avoid the
Copy code
Would you like to configure schedules for this deployment? [y/n] (y): n
prompt for the flows that have triggers (e.g. run B after A)?
m
✅ thought for 98.3 seconds
Yes — the easiest way is to make the deploy non-interactive or explicitly set 
schedules: []
for the triggered deployment. Why the prompt appears Triggers and schedules are independent. If a deployment has 
triggers
but no 
schedules
field, 
prefect deploy
may still ask whether you want to configure schedules interactively. Best fix in 
prefect.yaml
for Flow B Set an empty schedules list:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    schedules:
      - interval: 300

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.name: flow_a
That tells Prefect explicitly: this deployment has no schedules, so it should not prompt you to add one. Other ways to suppress prompts entirely 1. Use the global no-prompt flag:
Copy code
bash
prefect --no-prompt deploy
or for one deployment:
Copy code
bash
prefect --no-prompt deploy --name flow_b/prod
2. Set the environment variable:
Copy code
bash
export PREFECT_CLI_PROMPT=false
prefect deploy
Recommended approach If you're using 
prefect.yaml
, I’d recommend: * define schedules for scheduled deployments * use 
schedules: []
for trigger-only deployments * use 
prefect --no-prompt deploy
in CI/CD Docs: * Using prefect.yaml * Deployments concepts If you want, I can give you a complete 
prefect.yaml
example with: * 
flow_a/prod
scheduled every 
N
minutes * 
flow_b/prod
trigger-only with no schedule prompt
l
@Marvin How to automatically stop deployments that are not defined in prefect.yaml after running 
prefect deploy --all
?
m
✅ thought for 102.8 seconds
Short answer: you can’t do this automatically with 
prefect deploy --all
alone. 
prefect deploy --all
only creates/updates deployments defined in 
prefect.yaml
. It does not prune, disable, or delete deployments that were previously created but are no longer in the file. So if a deployment disappears from 
prefect.yaml
, it will generally remain in Prefect until you explicitly: * delete it, or * pause/clear its schedules, depending on what “stop” means for you --- What “stop” could mean 1. Stop scheduled runs but keep the deployment * clear or pause its schedules * useful if you want to preserve the deployment for manual runs/history 2. Completely remove the deployment * delete the deployment * useful if it is truly obsolete --- There is no built-in prune flag There isn’t a built-in equivalent of “deploy all and remove anything not in `prefect.yaml`”. So the usual pattern is: * run 
prefect deploy --all
* separately run a cleanup script that: * lists deployments currently in Prefect * compares them to the ones declared in 
prefect.yaml
* deletes or disables the extras --- Recommended approach If you want obsolete deployments to no longer run, the cleanest options are: * Delete them explicitly if they are gone for good * Keep them in 
prefect.yaml
with 
paused: true
 for a transition period Example:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    paused: false

  - name: old-prod
    entrypoint: old_flow.py:old_flow
    paused: true
That lets you manage deprecation declaratively before eventual deletion. --- CLI commands to use Delete a deployment:
Copy code
bash
prefect deployment delete flow_a/prod
If instead you only want to stop schedules, use the deployment schedule commands from the CLI for that deployment. --- Practical CI/CD pattern A common setup is:
Copy code
bash
prefect --no-prompt deploy --all
python cleanup_orphaned_deployments.py
Where `cleanup_orphaned_deployments.py`: * parses 
prefect.yaml
* builds the expected set of deployment names like 
flow-name/deployment-name
* queries existing deployments from Prefect * deletes anything not expected --- Bottom line No, 
prefect deploy --all
does not automatically stop deployments missing from 
prefect.yaml
. You need a separate cleanup step. Useful docs: * Using prefect.yaml * Deployments concepts If you want, I can draft a small Python cleanup script that deletes deployments not present in your 
prefect.yaml
.
l
@Marvin This didn't work, I triggered a 
quick run
on the deployment and 
flow_b
didn't start: triggers: - enabled: true match: prefect.resource.name: flow_b/prod expect: - prefect.flow-run.completed
m
✅ thought for 15.6 seconds
The issue is your trigger is matching the wrong resource. Your config says:
Copy code
yaml
match:
  prefect.resource.name: flow_b/prod
But if you want Flow B to run after Flow A completes, then the trigger on B must match events emitted by A, not B. So it should match Flow A's resource, not 
flow_b/prod
. Also, for a 
prefect.flow-run.completed
event, the primary resource is the flow run / flow, not the deployment name in the way you're using it. So the first fix is conceptually: * deployment for B has the trigger * trigger matches completion events from A Something more like:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    schedules:
      - interval: 300

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.name: flow_a
Important detail about 
quick run
 A quick run triggers a deployment run manually, but whether it emits exactly the event shape your trigger expects depends on what resource fields you are matching. If you match the wrong field/value, B will never trigger. Your current trigger is effectively saying: “Run B when B completes.” Not: “Run B when A completes.” --- Recommended correction Try this first:
Copy code
yaml
triggers:
  - enabled: true
    expect:
      - prefect.flow-run.completed
    match:
      prefect.resource.name: flow_a
If you want to be even more explicit, matching by resource id is often safer than name. --- Another important caveat: avoid self-triggering loops If B matches B's own completion event, it could create a loop in some setups. So definitely do not point B's trigger at 
flow_b/...
unless you intentionally want recursion. --- Why deployment name may not be the right matcher For 
prefect.flow-run.completed
, the event is about a flow run. The deployment may appear in related resources, but 
prefect.resource.name
is usually not the deployment full name you want to use here. So matching on:
Copy code
yaml
prefect.resource.name: flow_b/prod
is likely not matching any emitted completion event from A. --- What to do next I’d suggest: 1. Put the trigger on B 2. Match 
flow_a
, not 
flow_b/prod
3. Test by running A manually 4. Check the event/resource details if it still doesn’t fire A corrected sketch:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.name: flow_a
If you want, I can help you write the exact correct trigger filter for your case — for example matching: * by flow name * by deployment id/name * only successful runs of 
flow_a/prod
l
@Marvin Is it necessary a new deployment for the trigger? Could the trigger be defined in the deployment that triggers the run of a new flow? What if this triggered flow needs parameters?
m
✅ thought for 74.3 seconds
Yes — if you want a trigger to start Flow B, there must be a deployment for Flow B. In Prefect 3.x, deployment triggers run the deployment they are attached to. So: * a trigger on deployment A can only launch A * if you want B to run after A, the trigger should be attached to B's deployment, and it should listen for events from A So the pattern is: * Deployment A: scheduled every N minutes * Deployment B: has a trigger that watches for A's completion * When the trigger fires, B's deployment is run --- Can the trigger be defined on A and start B? Not as a deployment trigger in 
prefect.yaml
. Deployment triggers are shorthand for “create an automation that runs *this deployment*”. They do not say “when A happens, run some other deployment C”. So for your use case, the answer is: No, not as a deployment-level trigger in 
prefect.yaml
attached to A. Yes, if you instead use a more general automation outside that shorthand, but that’s a different mechanism. --- What if the triggered flow needs parameters? You can provide parameters on the trigger configuration for the deployment being triggered. So B's deployment can have:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    schedules:
      - interval: 300

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.name: flow_a
        parameters:
          x: 123
          mode: "from-a"
Then 
flow_b
could be:
Copy code
python
from prefect import flow

@flow
def flow_b(x: int, mode: str):
    print(x, mode)
When the trigger fires, Prefect runs B with those parameters. --- Important limitation on parameters Those trigger parameters are for the triggered deployment run. They are not automatically “the return value of A”. So if B needs data produced by A, you generally have three options: 1. Static parameters in the trigger * good for fixed config 2. Pass values derived from the event * depending on trigger/automation templating support 3. Store A's output externally * database, object storage, block-backed storage, etc. * then B reads it If B needs A's actual in-memory Python result directly, separate deployments are usually not the right pattern; using B as a subflow of A is simpler. --- Correct mental model Trigger attached to A: * “when X happens, run A” Trigger attached to B: * “when X happens, run B” So to say: “when A completes, run B” you put the trigger on B, and match completion events from A. --- Recommended shape
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    schedules:
      - interval: 300

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.name: flow_a
        parameters:
          source: "flow_a"
--- If you want, I can help you write the exact valid 
prefect.yaml
 for: * A scheduled every N minutes * B triggered after A * B receiving parameters and we can make sure the event matching is correct for your specific flow names.
l
@Marvin Could I make the trigger to match a deployment 
flow/deployment-name
instead of a flow?
m
✅ thought for 120.2 seconds
Yes — you can target a specific upstream deployment, but not by matching a single string like 
flow_a/prod
in 
prefect.resource.name
. For a 
prefect.flow-run.completed
event, the event is about a flow run, and the upstream deployment shows up as a related resource. So to distinguish 
flow_a/prod
from, say, 
flow_a/dev
, you should match the related resources for: * the flow named 
flow_a
* the deployment named 
prod
So the right approach is to use 
match_related
, not just 
match
. A good pattern is:
Copy code
yaml
deployments:
  - name: prod
    entrypoint: flow_a.py:flow_a
    schedules:
      - interval: 300

  - name: prod
    entrypoint: flow_b.py:flow_b
    schedules: []
    triggers:
      - type: event
        enabled: true
        expect:
          - prefect.flow-run.completed
        match:
          prefect.resource.id: prefect.flow-run.*
        match_related:
          - prefect.resource.role: flow
            prefect.resource.name: flow_a
          - prefect.resource.role: deployment
            prefect.resource.name: prod
That effectively says: when a flow run completes and the related flow is 
flow_a
 and the related deployment is 
prod
 then run this deployment (
flow_b/prod
) --- Important detail The deployment related resource name is typically just the deployment name part, e.g. 
prod
, not the combined 
flow_a/prod
. So this usually won’t be the right matcher:
Copy code
yaml
prefect.resource.name: flow_a/prod
Instead, split it into flow + deployment matching via 
match_related
. --- Even safer option If you want to avoid any ambiguity, you can match by the upstream deployment resource id instead of its name, if you know it. --- So in short: Can you target a specific deployment rather than only a flow? Yes. Can you do it by matching literal 
flow/deployment-name
as one resource name string? Usually no. Use 
match_related
with separate flow and deployment constraints. If you want, I can help turn your current 
prefect.yaml
into a fully working A→B example with: * 
flow_a/prod
scheduled * 
flow_b/prod
triggered only by 
flow_a/prod
* parameters passed into 
flow_b
2 Views
Open in Slack
PreviousNext
Powered by