https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> About prefect best practices, would i...
# ask-marvin
l
@Marvin About prefect best practices, would it be better to have a single flow that only runs other flows sequentially or have one deployment for each flow and make them run sequentially in some way? They don't depend in each other, the only reason to run them sequentially is to avoid hitting API quota limits.
m
✅ thought for 112.6 seconds
If the flows are independent, the best practice is usually: keep one deployment per flow and enforce the quota where the API is actually used, rather than building a “master sequential flow” just to serialize them. Why: - each flow stays independently deployable, runnable, retryable, and observable - you avoid coupling unrelated pipelines together - quota control becomes more robust, because it still works even if someone triggers flows manually, on a schedule, or from somewhere else Recommended approach Use: - one deployment per flow - plus a Prefect concurrency limit around the API-calling work That way, even if multiple flows are scheduled at once, only the allowed amount of API work happens concurrently. For Prefect 3.x, the most relevant mechanisms are: - global concurrency limits for shared external resources - tag-based task concurrency limits if the quota is task-level - deployment concurrency limits if you specifically want to limit runs of one deployment A “parent flow that runs child flows sequentially” is reasonable if you truly want one orchestrated pipeline and want the UI to show a parent/child relationship. But if the only reason is quota protection, it’s usually not the best design. Practical guidance Best option for your case: - Keep separate flows/deployments - Add a shared concurrency limit around the API calls This protects you no matter how the flows are triggered. Conceptually:
Copy code
python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def call_api(payload):
    with concurrency("external-api", occupy=1, strict=True):
        # do API request here
        ...
If all flows use that same limit, they’ll effectively serialize access to the API. When a parent flow makes sense Use a parent flow if you want: - a single button/schedule to run a whole sequence - explicit ordering for business reasons - one high-level run representing the full process In that case, you can have an orchestrator flow call deployments sequentially with 
run_deployment(...)
. But caveat: - if you rely only on a parent flow for sequencing, quota protection can be bypassed by other triggers or manual runs - retries and failures become a bit more coupled at the orchestration level So even with a parent flow, I’d still recommend enforcing concurrency at the API/task layer. What I would not do I would not make one giant flow whose only purpose is “run unrelated flows one after another” unless they’re logically one workflow. It tends to reduce flexibility without giving you stronger quota guarantees than concurrency controls do. Rule of thumb - Independent flows → separate deployments - Shared external quota → shared concurrency limit - Need one orchestrated sequence view → optional parent/orchestrator flow on top Relevant docs and references: - Run deployments from a flow - Global concurrency limits - Tag-based concurrency limits - Deployment concurrency limits If you want, I can show you: 1. a separate deployments + shared concurrency limit example, or 2. a parent orchestrator flow example, depending on how you currently trigger these flows.
l
@Marvin Show a separate deployments + shared concurrency limit example
m
✅ thought for 25.9 seconds
Absolutely — here’s a simple Prefect 3.x pattern with: - two independent flows - one deployment per flow - one shared global concurrency limit that serializes access to the external API This is the pattern I’d recommend for your use case. --- 1. Define the flows Both flows can run whenever they want, but the actual API section is guarded by the same concurrency limit name.
Copy code
from prefect import flow, task, get_run_logger
from prefect.concurrency.sync import concurrency
import time

@task
def call_external_api(flow_name: str, item: str):
    logger = get_run_logger()

    with concurrency("vendor-api", occupy=1, strict=True):
        <http://logger.info|logger.info>(f"{flow_name}: acquired API slot for {item}")
        time.sleep(5)  # simulate the API call
        <http://logger.info|logger.info>(f"{flow_name}: finished API call for {item}")

@flow
def sync_customers():
    for customer_id in ["cust-1", "cust-2", "cust-3"]:
        call_external_api(customer_id, customer_id)

@flow
def sync_orders():
    for order_id in ["order-1", "order-2", "order-3"]:
        call_external_api("sync_orders", order_id)
Small correction: the first flow should pass its flow name too. Here’s the cleaned-up version:
Copy code
from prefect import flow, task, get_run_logger
from prefect.concurrency.sync import concurrency
import time

@task
def call_external_api(flow_name: str, item: str):
    logger = get_run_logger()

    with concurrency("vendor-api", occupy=1, strict=True):
        <http://logger.info|logger.info>(f"{flow_name}: acquired API slot for {item}")
        time.sleep(5)  # simulate API call
        <http://logger.info|logger.info>(f"{flow_name}: finished API call for {item}")

@flow
def sync_customers():
    for customer_id in ["cust-1", "cust-2", "cust-3"]:
        call_external_api("sync_customers", customer_id)

@flow
def sync_orders():
    for order_id in ["order-1", "order-2", "order-3"]:
        call_external_api("sync_orders", order_id)
Because both flows use 
concurrency("vendor-api", ...)
, only one of those protected sections can run at a time across all flow runs using that same limit. --- 2. Create the shared global concurrency limit Create it once:
Copy code
prefect gcl create vendor-api --limit 1
That gives you one shared slot for the resource named 
vendor-api
. If you later decide the API can tolerate 2 concurrent requests, you can raise the limit to 
2
. Docs: - Global concurrency limits --- 3. Deploy each flow separately You can deploy each flow as its own deployment. A simple Python-based deployment example:
Copy code
from flows import sync_customers, sync_orders

if __name__ == "__main__":
    sync_customers.deploy(
        name="customers-prod",
        work_pool_name="my-pool",
    )

    sync_orders.deploy(
        name="orders-prod",
        work_pool_name="my-pool",
    )
Or if you keep them in separate files, each can deploy independently. You can also use the CLI:
Copy code
prefect deploy --help
Docs: - Deploy via Python - Deploy with prefect.yaml --- 4. What happens at runtime Suppose: - 
sync_customers
starts at 10:00 - 
sync_orders
starts at 10:01 Both flows may be running at the same time overall, but only one task inside 
with concurrency("vendor-api")
will hold the slot at once. So in practice: - 
sync_customers
gets the slot first - 
sync_orders
waits before entering the protected API section - when the slot is released, the next waiting task proceeds This gives you serialized API usage without forcing the entire flows to be serialized. That’s usually better than “run whole flow A, then whole flow B”, because: - non-API work can still overlap - each flow remains independent - quota protection is enforced centrally --- 5. If you want to serialize the whole flow, not just the API call
If the entire flow should never overlap because every part of it contributes to the quota, you can wrap the whole body:
Copy code
from prefect import flow, get_run_logger
from prefect.concurrency.sync import concurrency
import time

@flow
def sync_customers():
    logger = get_run_logger()

    with concurrency("vendor-api", occupy=1, strict=True):
        <http://logger.info|logger.info>("sync_customers starting")
        time.sleep(10)
        <http://logger.info|logger.info>("sync_customers finished")

@flow
def sync_orders():
    logger = get_run_logger()

    with concurrency("vendor-api", occupy=1, strict=True):
        <http://logger.info|logger.info>("sync_orders starting")
        time.sleep(10)
        <http://logger.info|logger.info>("sync_orders finished")
That’s valid too — it just reduces parallelism more aggressively. My rule of thumb: - wrap only the API-calling section if possible - wrap the whole flow only if the whole flow should count against the quota --- 6. Alternative: tag-based concurrency If the quota is specifically around Prefect tasks, another nice option is to tag those tasks and limit the tag. Example task:
Copy code
from prefect import task

@task(tags=["vendor-api"])
def call_external_api(item: str):
    ...
Then create a concurrency limit for that tag. But for your use case, the explicit 
concurrency("vendor-api")
context manager is often clearer and more flexible, especially if the protected code is not just a single Prefect task call. Docs: - Tag-based concurrency limits --- 7. Recommended final structure I’d structure it like this: - 
sync_customers
flow → deployment 
customers-prod
- 
sync_orders
flow → deployment 
orders-prod
- shared limit name: 
vendor-api
- schedules/manual triggers/automations can trigger either deployment independently - shared concurrency limit ensures quota safety That gives you: - independent deployments - independent retries - simpler operations - reliable quota protection --- If you want, I can also show: - a complete repo layout for this pattern, or - the same example using 
prefect.yaml
 instead of Python deployment code.
l
@Marvin How to define the same in a yaml?
m
✅ thought for 74.9 seconds
Yes — for separate deployments + shared concurrency limit, the 
prefect.yaml
defines the deployments, while the shared global concurrency limit is still created separately and referenced in code. So the split is: - 
prefect.yaml
→ defines 
sync_customers
and 
sync_orders
deployments - Python code → uses 
with concurrency("vendor-api", ...)
- CLI/UI/API → creates the global limit 
vendor-api
--- Example project structure
Copy code
.
├── flows.py
└── prefect.yaml
--- flows.py
Copy code
from prefect import flow, task, get_run_logger
from prefect.concurrency.sync import concurrency
import time

@task
def call_external_api(flow_name: str, item: str):
    logger = get_run_logger()

    with concurrency("vendor-api", occupy=1, strict=True):
        <http://logger.info|logger.info>(f"{flow_name}: acquired API slot for {item}")
        time.sleep(5)  # simulate API call
        <http://logger.info|logger.info>(f"{flow_name}: finished API call for {item}")

@flow
def sync_customers():
    for customer_id in ["cust-1", "cust-2", "cust-3"]:
        call_external_api("sync_customers", customer_id)

@flow
def sync_orders():
    for order_id in ["order-1", "order-2", "order-3"]:
        call_external_api("sync_orders", order_id)
--- prefect.yaml Here’s a minimal example with two deployments:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

deployments:
  - name: customers-prod
    entrypoint: flows.py:sync_customers
    work_pool:
      name: my-pool

  - name: orders-prod
    entrypoint: flows.py:sync_orders
    work_pool:
      name: my-pool
That’s enough for two separate deployments. Then deploy them with:
Copy code
prefect deploy --all
or just one:
Copy code
prefect deploy --name customers-prod
prefect deploy --name orders-prod
CLI reference: - `prefect deploy` - prefect.yaml guide --- Create the shared global concurrency limit This part is not defined in 
prefect.yaml
. Create it once with:
Copy code
prefect gcl create vendor-api --limit 1
Docs: - Global concurrency limits --- Optional: also limit each deployment’s concurrent runs If you also want to prevent multiple runs of the same deployment from piling up, you can add 
concurrency_limit
to each deployment. Example:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

deployments:
  - name: customers-prod
    entrypoint: flows.py:sync_customers
    concurrency_limit: 1
    work_pool:
      name: my-pool

  - name: orders-prod
    entrypoint: flows.py:sync_orders
    concurrency_limit: 1
    work_pool:
      name: my-pool
That means: - only one 
customers-prod
run at a time - only one 
orders-prod
run at a time - and because both flows use 
concurrency("vendor-api")
, only one protected API section across both flows at a time Important distinction: - 
concurrency_limit
in YAML = per-deployment run limit - 
concurrency("vendor-api")
= shared cross-flow resource limit If your actual concern is API quota, the shared global limit is the important part. --- If you want schedules too You can also attach schedules in 
prefect.yaml
, for example:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

deployments:
  - name: customers-prod
    entrypoint: flows.py:sync_customers
    work_pool:
      name: my-pool
    schedule:
      cron: "0 * * * *"
      timezone: "UTC"

  - name: orders-prod
    entrypoint: flows.py:sync_orders
    work_pool:
      name: my-pool
    schedule:
      cron: "30 * * * *"
      timezone: "UTC"
Even if both schedules overlap, the shared 
vendor-api
limit still protects the API. --- Bottom line For your case, I’d use: 1. 
prefect.yaml
with one deployment per flow 2. 
with concurrency("vendor-api", occupy=1, strict=True)
in the code
3. 
prefect gcl create vendor-api --limit 1
If you want, I can give you a copy-pasteable full example with: - schedules - parameters - and a worker startup command too.
l
@Marvin Let's say I have 4 flow runs that were late, how could I make sure that only the last one is ran instead of all of them?
m
✅ thought for 135.6 seconds
Short answer: Prefect doesn’t have a built-in “latest only” switch for scheduled backlog runs. So if 4 scheduled runs are late, Prefect will not automatically say “drop the first 3 and run only the newest one” unless you add some control logic. What Prefect can do natively 1. Limit concurrency to 1 So only one run executes at a time. 2. Choose what happens when a new run arrives while one is active With deployment concurrency: - 
ENQUEUE
→ queue it - 
CANCEL_NEW
→ cancel the new one But note: 
CANCEL_NEW
gives you “keep the old one, drop the new one”, which is the opposite of “latest only”. So for your exact requirement, latest only, you usually need custom cleanup logic. --- Best practical pattern Use: - deployment concurrency limit = 1 - plus a small “startup guard” at the beginning of the flow that checks for newer scheduled runs and exits/cancels itself if it is not the latest This works well because: - only one run executes at a time - older delayed runs can self-skip when they finally start - the newest run is the one that actually does the work --- Example approach Let’s say you have one deployment called 
customers-prod
. At the top of the flow, check: - what deployment this run belongs to - whether there is a newer scheduled/running/pending run for the same deployment - if yes, this run should exit early Conceptually:
Copy code
from prefect import flow, get_run_logger
from prefect.runtime import flow_run
from prefect.client.orchestration import get_client

async def should_this_run_continue() -> bool:
    current_run_id = flow_run.id
    deployment_id = flow_run.deployment_id

    async with get_client() as client:
        runs = await client.read_flow_runs(
            # filter to same deployment
            # sort newest first
        )

    # if there's a newer run for same deployment, return False
    return True

@flow
async def sync_customers():
    logger = get_run_logger()

    if not await should_this_run_continue():
        <http://logger.info|logger.info>("Skipping this run because a newer run exists.")
        return

    <http://logger.info|logger.info>("This is the latest run, proceeding.")
    # do real work here
That’s the core pattern. --- Even simpler operational pattern If you don’t want code-based self-skipping, another option is a separate cleanup process that: - looks for scheduled late runs for a deployment - keeps only the newest one - cancels the older ones That can run: - on a schedule - before triggering work - or via automation --- What I recommend for your case If your goal is “data freshness matters more than processing every interval”: - set deployment concurrency limit to 1 - add self-skip logic so old runs abandon themselves if a newer run exists That gives you the effective behavior of: - backlog may exist briefly - but only the latest useful run does real work --- What this looks like in 
prefect.yaml
 You can at least add the concurrency limit there:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

deployments:
  - name: customers-prod
    entrypoint: flows.py:sync_customers
    concurrency_limit: 1
    work_pool:
      name: my-pool
This alone does not give “latest only”, but it prevents overlap. Docs: - Deployment concurrency limits --- Important nuance There are two opposite policies: - oldest wins → process backlog in order - latest wins → skip stale runs, process only freshest state Prefect’s built-in concurrency controls are better aligned with oldest wins. For latest wins, you need custom logic. --- Recommended implementation options Option A: self-skip in flow code Best when: - you control the flow code - you want the simplest durable pattern Option B: external cleanup/cancellation flow Best when: - you want to cancel stale runs before they start
- you want centralized backlog management Option C: redesign schedule/triggering model If only the latest state matters, sometimes event-driven triggering or less frequent schedules are better than allowing backlog to accumulate. --- If you want, I can give you a full Prefect 3 example of the self-skip latest-only pattern that: - checks for newer runs in the same deployment - logs why it is skipping - and works with an async flow.
l
@Marvin Many of these deployments have the same work pool, entrypoint, schedule, and similar name, except that the parameters are different. Is there a way of re-using a lot of the YAML so to make it more organized and easy to maintain?
m
✅ thought for 120.0 seconds
Yes — definitely. The best way to reduce duplication in 
prefect.yaml
is to use standard YAML anchors and merge keys. Prefect reads 
prefect.yaml
as normal YAML, so you can reuse shared pieces like: - 
entrypoint
- 
work_pool
- 
schedule
- tags - descriptions - even common parameter subsets This is usually the cleanest way to manage many similar deployments. --- Recommended pattern: base deployment config + per-deployment overrides Example:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

definitions:
  common_work_pool: &common_work_pool
    name: my-pool

  common_schedule: &common_schedule
    cron: "0 * * * *"
    timezone: "UTC"

  base_deployment: &base_deployment
    entrypoint: flows.py:sync_data
    work_pool: *common_work_pool
    schedule: *common_schedule
    tags:
      - api
      - production

deployments:
  - <<: *base_deployment
    name: sync-customers
    parameters:
      dataset: customers
      endpoint: /customers

  - <<: *base_deployment
    name: sync-orders
    parameters:
      dataset: orders
      endpoint: /orders

  - <<: *base_deployment
    name: sync-products
    parameters:
      dataset: products
      endpoint: /products
That way: - all shared config lives in one place - each deployment only defines what is different - maintenance gets much easier --- If you need slightly different schedules or work pools You can define multiple reusable blocks:
Copy code
prefect-version: "3.0.0"

name: quota-demo

build: null
push: null
pull: null

definitions:
  pool_default: &pool_default
    name: my-pool

  hourly_schedule: &hourly_schedule
    cron: "0 * * * *"
    timezone: "UTC"

  daily_schedule: &daily_schedule
    cron: "0 2 * * *"
    timezone: "UTC"

  base_deployment: &base_deployment
    entrypoint: flows.py:sync_data
    work_pool: *pool_default
    tags:
      - api
      - production

deployments:
  - <<: *base_deployment
    name: sync-customers
    schedule: *hourly_schedule
    parameters:
      dataset: customers

  - <<: *base_deployment
    name: sync-orders
    schedule: *hourly_schedule
    parameters:
      dataset: orders

  - <<: *base_deployment
    name: sync-products
    schedule: *daily_schedule
    parameters:
      dataset: products
--- Important caveat: nested YAML merges are shallow If you override a nested object like 
work_pool.job_variables
, you may replace the whole nested mapping unless you also anchor that nested part separately. For example, better:
Copy code
definitions:
  common_job_vars: &common_job_vars
    env: prod
    region: us-east-1

  common_work_pool: &common_work_pool
    name: my-pool
    job_variables:
      <<: *common_job_vars

deployments:
  - name: sync-customers
    entrypoint: flows.py:sync_data
    work_pool:
      <<: *common_work_pool
      job_variables:
        <<: *common_job_vars
        memory: 2Gi
    parameters:
      dataset: customers
If you don’t do that, overriding 
job_variables
can wipe out inherited nested keys. --- Can Prefect variables help too? Yes, but for a slightly different purpose. Use YAML anchors for: - structural reuse - keeping the file organized Use Prefect variables or environment variables for: - values that change by environment - values you don’t want hardcoded Example:
Copy code
definitions:
  base_deployment: &base_deployment
    entrypoint: flows.py:sync_data
    work_pool:
      name: "{{ prefect.variables.default_work_pool }}"
That said, for your specific case — many deployments with the same structure and different parameters — anchors are the main tool. --- Best practice for your scenario If all deployments have: - same flow function / entrypoint - same work pool - same schedule - similar names - only different parameters then I’d strongly recommend: 1. one shared 
base_deployment
anchor 2. optional separate anchors for 
work_pool
and 
schedule
3. each deployment only setting 
name
and 
parameters
That’s the cleanest setup. --- Example tailored exactly to your case
Copy code
prefect-version: "3.0.0"

name: vendor-sync

build: null
push: null
pull: null

definitions:
  common_schedule: &common_schedule
    cron: "*/30 * * * *"
    timezone: "UTC"

  common_work_pool: &common_work_pool
    name: my-pool

  base_deployment: &base_deployment
    entrypoint: flows.py:sync_vendor_resource
    work_pool: *common_work_pool
    schedule: *common_schedule
    concurrency_limit: 1
    tags:
      - vendor-api

deployments:
  - <<: *base_deployment
    name: vendor-sync-customers
    parameters:
      resource: customers
      path: /customers

  - <<: *base_deployment
    name: vendor-sync-orders
    parameters:
      resource: orders
      path: /orders

  - <<: *base_deployment
    name: vendor-sync-products
    parameters:
      resource: products
      path: /products

  - <<: *base_deployment
    name: vendor-sync-invoices
    parameters:
      resource: invoices
      path: /invoices
This is probably very close to what you want. Docs: - prefect.yaml guide If you want, I can help you turn your current 
prefect.yaml
 into a DRY version using anchors.
l
@Marvin How to reuse something like 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: {custom} - prefect.resource.role: deployment prefect.resource.name: {custom}
m
✅ thought for 16.6 seconds
Yes — you can reuse that too with YAML anchors, but because part of it changes per deployment (
{custom}
), the cleanest pattern is: - anchor the common trigger structure - override just the 
match_related
values per deployment One important note: YAML itself does not do string substitution like 
{custom}
automatically. So you can’t define one block with a placeholder and expect YAML to replace it per deployment. That means the reusable part is the static structure, and the per-deployment names still need to be filled in explicitly. --- Recommended pattern
Copy code
prefect-version: "3.0.0"

name: my-project

definitions:
  completed_flow_trigger_base: &completed_flow_trigger_base
    type: event
    enabled: true
    expect:
      - prefect.flow-run.Completed
    match:
      prefect.resource.id: prefect.flow-run.*

deployments:
  - name: deployment-a
    entrypoint: flows.py:my_flow
    schedules: []
    triggers:
      - <<: *completed_flow_trigger_base
        match_related:
          - prefect.resource.role: flow
            prefect.resource.name: upstream-flow-a
          - prefect.resource.role: deployment
            prefect.resource.name: upstream-deployment-a

  - name: deployment-b
    entrypoint: flows.py:my_flow
    schedules: []
    triggers:
      - <<: *completed_flow_trigger_base
        match_related:
          - prefect.resource.role: flow
            prefect.resource.name: upstream-flow-b
          - prefect.resource.role: deployment
            prefect.resource.name: upstream-deployment-b
That gives you reuse for: - 
type
- 
enabled
- 
expect
- 
match
and keeps only the deployment-specific part repeated. --- If you want to reuse the entire trigger except one nested field Because YAML merges are shallow, you usually need to split nested parts into their own anchors if you want finer control. For example:
Copy code
prefect-version: "3.0.0"

name: my-project

definitions:
  trigger_match: &trigger_match
    prefect.resource.id: prefect.flow-run.*

  completed_trigger: &completed_trigger
    type: event
    enabled: true
    expect:
      - prefect.flow-run.Completed
    match:
      <<: *trigger_match

deployments:
  - name: deployment-a
    entrypoint: flows.py:my_flow
    schedules: []
    triggers:
      - <<: *completed_trigger
        match_related:
          - prefect.resource.role: flow
            prefect.resource.name: flow-a
          - prefect.resource.role: deployment
            prefect.resource.name: deployment-a
This is more verbose up front, but easier to maintain if you have lots of triggers. --- Can you reuse 
match_related
too? Only if it is exactly the same across deployments. Example:
Copy code
definitions:
  common_match_related: &common_match_related
    - prefect.resource.role: flow
      prefect.resource.name: same-flow
    - prefect.resource.role: deployment
      prefect.resource.name: same-deployment
Then:
Copy code
triggers:
  - <<: *completed_trigger
    match_related: *common_match_related
But in your example, those names are the part that changes, so that piece usually can’t be shared directly. --- If the only thing changing is a name pair, a practical pattern is this ``` definitions: event_completed_trigger: &event_completed_trigger type: event enabled: true expect: - prefect.flow-run.Completed match: prefect.resource.id: prefect.flow-run.* deployments: - name: downstream-customers entrypoint: flows.py:downstream_flow schedules: [] triggers: - <<: *event_completed_trigger match_related: - prefect.resource.role: flow prefect.resource.name: upstream-customers - prefect.resource.role: deployment prefect.resource.name: upstream-customers-deployment - name: downstream-orders entrypoint: flows.py:downstream_flow schedules: [] triggers: - <<: *event_completed_trigger match_related:
- prefect.resource.role: flow prefect.resource.name: upstream-orders - prefect.resource.role: deployment prefect.resource.name: upstream-orders-deployment
Copy code
---

*Important limitation*

If you were hoping for something like:
prefect.resource.name: {custom} ``` and then have Prefect fill 
{custom}
from the deployment name, Prefect/YAML does not support that kind of templated substitution in arbitrary YAML fields. So: - anchors/aliases → yes - placeholder replacement per deployment → not natively in plain 
prefect.yaml
If you need that level of generation, it’s often better to: - generate 
prefect.yaml
from another tool, or - define deployments in Python instead of hand-authoring many similar YAML entries --- My recommendation For a manageable number of deployments: - use an anchor for the common trigger body - repeat only 
match_related
For a large number of deployments: - consider generating the deployment config in Python instead of maintaining lots of repeated YAML Docs: - prefect.yaml guide If you want, I can show you: 1. a full YAML example with reusable trigger anchors, or 2. a Python deployment-generation example that avoids repeating all those trigger blocks.
l
@Marvin Could match_related match multiple deployments of the same flow? So a run could be triggered if either of those deloyments are completed?
m
✅ thought for 109.2 seconds
Yes — if any of several deployments should trigger it, you can do that. But the key detail is: - 
match_related
as a list of separate objects behaves like AND - to get OR, put multiple values in one field inside a single 
match_related
object So this is the wrong shape for “either deployment A or B”:
Copy code
match_related:
  - prefect.resource.role: deployment
    prefect.resource.name: deployment-a
  - prefect.resource.role: deployment
    prefect.resource.name: deployment-b
That effectively means: match an event related to deployment-a AND deployment-b, which is not what you want. --- Use this shape instead
Copy code
match_related:
  prefect.resource.role: deployment
  prefect.resource.name:
    - deployment-a
    - deployment-b
That gives you OR across those names. --- Applied to your case If you want a downstream deployment to trigger when either of two upstream deployments of the same flow completes:
Copy code
triggers:
  - type: event
    enabled: true
    expect:
      - prefect.flow-run.Completed
    match:
      prefect.resource.id: prefect.flow-run.*
    match_related:
      prefect.resource.role: deployment
      prefect.resource.name:
        - upstream-deployment-1
        - upstream-deployment-2
That should match a completed flow run whose related deployment is either of those two deployments. --- If you also want to constrain by flow name You can still use 
match_related
, but remember the semantics. If you need: - related flow name = 
my-flow
- and related deployment name is one of 
[dep-a, dep-b]
then this is a valid AND across different fields inside the same object:
Copy code
match_related:
  prefect.resource.role:
    - flow
    - deployment
  prefect.resource.name:
    - my-flow
    - dep-a
    - dep-b
But that is usually too ambiguous because it doesn’t pair role-to-name cleanly. So in practice, if you need more precise logic like: - flow is 
my-flow
- deployment is one of 
dep-a
or 
dep-b
the clearer option is often a compound trigger or to just match on deployment names if those uniquely identify the upstream source. --- Best practical recommendation If the deployment names are unique and stable, use only deployment matching:
Copy code
match_related:
  prefect.resource.role: deployment
  prefect.resource.name:
    - upstream-customers-prod
    - upstream-customers-backfill
That is the cleanest “trigger if any of these deployments complete” pattern. --- Important caveat This only works if the 
prefect.flow-run.Completed
event includes the related deployment resource, which flow-run completion events normally do. --- Docs: - Event triggers - prefect.yaml guide If you want, I can show you the exact YAML anchor pattern to reuse a trigger where only the list of upstream deployment names changes.
l
@Marvin Give an example of a compound trigger with flow_a & deployment_a or flow_b & deployment_b use-case (two conditions of flow-deployment pairs). Also, would it be better to deploy using python to avoid a lot of the repetition of the yaml or more flexibility?
m
✅ thought for 96.5 seconds
Yes — and for that exact case, I’d actually lean toward Python deployment definitions if you have many similar trigger pairings. Because once you get into: - repeated trigger structures - repeated 
(flow_x, deployment_x)
pairs - OR/AND nesting - lots of near-identical deployments Python becomes much easier to maintain than large repetitive YAML. ## First: important modeling note For your use case, you usually do not want: - one event trigger matching the flow - and another event trigger matching the deployment as separate sibling event triggers against the same completion event. Why? Because a single 
prefect.flow-run.Completed
event already includes related resources like: - the flow - the deployment So the cleanest expression of “flow_a & deployment_a” is usually one event trigger that matches: - the event type - and the related flow - and the related deployment Then you OR that with another event trigger for 
(flow_b & deployment_b)
. So the logic is: 
(Completed event with related flow_a + deployment_a) OR (Completed event with related flow_b + deployment_b)
That maps nicely to a compound trigger with 
require: any
. --- ## YAML example Here’s a 
prefect.yaml
example for: - trigger downstream if flow_a from deployment_a completes - OR if flow_b from deployment_b completes
Copy code
prefect-version: "3.0.0"

name: trigger-demo

build: null
push: null
pull: null

deployments:
  - name: downstream
    entrypoint: flows.py:downstream_flow
    work_pool:
      name: my-pool
    triggers:
      - type: compound
        require: any
        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: deployment_a

          - 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_b
              - prefect.resource.role: deployment
                prefect.resource.name: deployment_b
### Why this works Each inner 
event
trigger means: - event is a completed flow run - AND related flow name matches - AND related deployment name matches The outer 
compound
with 
require: any
means: - trigger if either pair matches --- ## Reusing the common parts in YAML You can reduce repetition with anchors:
Copy code
prefect-version: "3.0.0"

name: trigger-demo

definitions:
  completed_flow_run_trigger: &completed_flow_run_trigger
    type: event
    enabled: true
    expect:
      - prefect.flow-run.Completed
    match:
      prefect.resource.id: prefect.flow-run.*

deployments:
  - name: downstream
    entrypoint: flows.py:downstream_flow
    work_pool:
      name: my-pool
    triggers:
      - type: compound
        require: any
        triggers:
          - <<: *completed_flow_run_trigger
            match_related:
              - prefect.resource.role: flow
                prefect.resource.name: flow_a
              - prefect.resource.role: deployment
                prefect.resource.name: deployment_a

          - <<: *completed_flow_run_trigger
            match_related:
              - prefect.resource.role: flow
                prefect.resource.name: flow_b
              - prefect.resource.role: deployment
                prefect.resource.name: deployment_b
This is probably the cleanest YAML version. --- ## Python example If you have many of these pairings, Python gets much nicer. ``` from prefect import flow @flow def downstream_flow(): print("Triggered") if name == "__main__": downstream_flow.deploy(
name="downstream", work_pool_name="my-pool", triggers=[ { "type": "compound", "require": "any", "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": "deployment_a", }, ], }, { "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_b", }, { "prefect.resource.role": "deployment", "prefect.resource.name": "deployment_b", }, ], }, ], } ], )
Copy code
---

## Better Python pattern: helper function

This is where Python starts to win.
from prefect import flow def completion_pair_trigger(flow_name: str, deployment_name: str): return { "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_name, }, { "prefect.resource.role": "deployment", "prefect.resource.name": deployment_name, }, ], } @flow def downstream_flow(): print("Triggered") if name == "__main__": upstream_pairs = [ ("flow_a", "deployment_a"), ("flow_b", "deployment_b"), ] downstream_flow.deploy( name="downstream", work_pool_name="my-pool", triggers=[ { "type": "compound", "require": "any", "triggers": [ completion_pair_trigger(flow_name, deployment_name) for flow_name, deployment_name in upstream_pairs ], } ], ) ``` That’s much easier to scale. --- ## YAML vs Python: which is better here? ### Stay with 
prefect.yaml
if: - you have a small to medium number of deployments - the config is mostly static - you want everything in one declarative file - YAML anchors are enough to keep it manageable ### Prefer Python if: - you have many similar deployments - you have repeated trigger logic - names/parameters/pairs are naturally represented as lists or dictionaries - you want loops, helper functions, validation, or generated config - you expect this setup to grow For your described situation, I’d say: If repetition is already becoming painful, Python is probably the better long-term choice. Especially because you’ve got: - many similar deployments - similar names - shared schedules/work pools/entrypoints - repeated trigger structures
- only certain values changing That is exactly the kind of thing Python handles better than YAML. --- ## My practical recommendation - If you only have a handful of deployments: use 
prefect.yaml
+ anchors - If you have dozens, or expect frequent changes: use Python deployment generation A very common pattern is: - store the varying pieces in a list of configs - loop over them in Python - call 
.deploy()
for each That gives you DRY config without wrestling YAML. --- ## One more caveat When using 
match_related
, the event must actually include those related resources. For 
prefect.flow-run.Completed
, matching related 
flow
and 
deployment
is the intended pattern. --- Useful references: - Event triggers - prefect.yaml guide - Deploy via Python If you want, I can next show you a fully DRY Python deployment-generator for: - one flow - many parameter sets - many trigger pairs - shared work pool/schedule/concurrency settings.
l
@Marvin How could I define this from the .yaml pull: - prefect.deployments.steps.set_working_directory: directory: /app In the python deployment for all deployments? Is it necessary?
m
✅ thought for 121.6 seconds
Short answer: you usually do not define that explicitly in Python, and it is not always necessary. That YAML:
Copy code
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /app
is mainly a deployment pull step telling the runtime where the code lives before loading the flow entrypoint. ## In Python, the answer depends on how you deploy ### 1. If you use 
flow.from_source(...).deploy(...)
Prefect usually handles the pull step for you automatically based on the source. Examples: - local path source → generates a 
set_working_directory
-style pull step - GitHub/repo source → generates a 
git_clone
pull step - remote storage source → generates the appropriate pull step So for this style, you typically do not need to manually recreate:
Copy code
prefect.deployments.steps.set_working_directory
Example:
Copy code
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.from_source(
        source="/app",
        entrypoint="flows.py:my_flow",
    ).deploy(
        name="my-deployment",
        work_pool_name="my-pool",
    )
In that case, Prefect understands the source location and handles the equivalent pull behavior. --- ### 2. If you use plain 
flow.deploy(...)
Then there is not a direct Python argument like:
Copy code
pull=[{"prefect.deployments.steps.set_working_directory": ...}]
in the same simple way you express it in 
prefect.yaml
. Usually with Python deployment APIs, you instead choose one of these patterns: - use 
from_source(...)
 so Prefect manages code retrieval / working directory - bake the code into the Docker image - ensure the worker environment already starts in the correct directory --- ## Is it necessary? ### Not necessary when: - your flow code is already available in the image/container - your Docker image sets 
WORKDIR /app
- your worker runs with the correct working directory - you use 
from_source(...)
and Prefect handles it ### Necessary when: - the runtime needs help locating code - the code is mounted or present in a specific directory like 
/app
- you rely on deployment pull steps rather than image packaging or 
from_source(...)
--- ## Practical recommendation ### If you are moving away from repetitive YAML The cleanest Python-based approach is often: - put your code in the image or - use 
from_source(...)
### Best option A: code baked into image If your worker image contains your code under 
/app
, then just make sure the image has:
Copy code
WORKDIR /app
and your flow entrypoint matches that layout. Then you usually don’t need a 
set_working_directory
pull step at all. ### Best option B: use 
from_source(...)
If code is not baked into the image, use:
Copy code
my_flow.from_source(
    source="/app",
    entrypoint="flows.py:my_flow",
).deploy(...)
or a Git source:
Copy code
my_flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="flows.py:my_flow",
).deploy(...)
This is the Python-native way to express where code comes from. --- ## If all deployments need 
/app
That’s often a sign that: - 
/app
should be your container 
WORKDIR
, or - 
/app
should be the 
source
root in 
from_source(...)
instead of repeatedly defining a pull step. --- ## Rule of thumb - Using 
prefect.yaml
with pull steps? 
set_working_directory
may be appropriate. - Using Python deployment definitions? Prefer 
from_source(...)
or container 
WORKDIR
rather than trying to manually replicate the YAML pull step everywhere. --- ## Example: many deployments in Python, all using 
/app
``` from prefect import flow @flow def sync_data(resource: str): print(f"syncing {resource}") if name == "__main__": resources = ["customers", "orders", "products"] for resource in resources: sync_data.from_source( source="/app", entrypoint="flows.py:sync_data", ).deploy( name=f"sync-{resource}",
work_pool_name="my-pool", parameters={"resource": resource}, ) ``` That’s the Python equivalent pattern I’d recommend. --- Relevant docs: - Deploy via Python - Store flow code / source patterns - prefect.yaml guide If you want, I can show you a side-by-side conversion of one of your YAML deployments into the recommended Python 
from_source(...).deploy(...)
form.
l
@Marvin I have this in a prefect-worker container volumes: - ./:/app working_dir: /app Then why do I receive this error? ValueError: Either an image or remote storage location must be provided when deploying a deployment.
m
✅ thought for 14.3 seconds
working_dir: /app
in your worker container only affects where the worker starts inside the container. It does not tell Prefect how a deployment should package or locate flow code when you call 
.deploy()
. That error:
Copy code
ValueError: Either an image or remote storage location must be provided when deploying a deployment.
means Prefect is saying:
“I can’t create this deployment unless you either:
- give me an image, or
- give me a *source/remote storage location*”
## Why your bind mount is not enough This:
Copy code
volumes:
  - ./:/app
working_dir: /app
makes the code available to that container at runtime. But 
.deploy()
needs deployment metadata that tells Prefect/worker how the code will be available when the deployment runs. A local Docker bind mount is: - local to your compose setup - not portable deployment metadata - not something Prefect records as deployment storage/source So Prefect does not infer: - “ah, code is mounted at 
/app
, so use that” You still need to deploy using one of these supported models: ### Option 1: provide an image Example:
Copy code
my_flow.deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    image="my-repo/my-image:latest",
)
### Option 2: provide a source location Example:
Copy code
my_flow.from_source(
    source="/app",
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
)
or Git:
Copy code
my_flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
)
--- ## Most likely what happened You probably did something like:
Copy code
my_flow.deploy(
    name="x",
    work_pool_name="my-pool",
)
without: - 
image=...
, and without - 
from_source(...)
For many work pools, Prefect requires one of those because the deployment must know where code comes from. --- ## What to do in your setup Since your worker container already mounts the repo at 
/app
, the simplest fix is usually:
Copy code
my_flow.from_source(
    source="/app",
    entrypoint="path/to/file.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
)
That tells Prefect explicitly: - source root is 
/app
- flow entrypoint is relative to that source ## Example
Copy code
from prefect import flow

@flow
def sync_customers():
    print("hello")

if __name__ == "__main__":
    sync_customers.from_source(
        source="/app",
        entrypoint="flows.py:sync_customers",
    ).deploy(
        name="customers-prod",
        work_pool_name="my-pool",
    )
--- ## Important caveat Using 
source="/app"
works if: - the deployment code runs in an environment where 
/app
exists - and the worker that executes the run also has that same mounted path So this is okay for: - local Docker Compose - controlled self-hosted environments But it is less portable than: - baking code into an image, or - using Git/remote storage --- ## Best long-term recommendation For local/self-hosted Docker: - 
from_source(source="/app", ...)
is fine For more portable/reliable deployments: - use a Docker image with the code baked in, or - use a Git repo source --- ## Summary Your mount and 
working_dir
affect the container runtime only. They do not satisfy Prefect’s requirement that a deployment specify where its code comes from. So fix it by using either: ### Local mounted source
Copy code
flow.from_source(source="/app", entrypoint="flows.py:my_flow").deploy(...)
### Or an image
Copy code
flow.deploy(image="my-image:latest", ...)
Docs: - Deploy via Python - Store flow code If you want, paste your current 
.deploy(...)
code and I can rewrite it into the correct 
from_source(...).deploy(...)
form.
l
@Marvin Where does the code run when I run prefect deploy --all with a yaml with pull: - prefect.deployments.steps.set_working_directory: directory: /app How to replicate the same with prefect deploy.py?
m
✅ thought for 13.4 seconds
prefect deploy --all
runs the deployment registration step wherever you execute the CLI. The 
pull:
section is not executed during `prefect deploy`; it is stored on the deployment and executed later by the worker when a flow run starts. So with:
Copy code
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /app
the sequence is: 1. you run 
prefect deploy --all
somewhere 2. Prefect reads 
prefect.yaml
and registers the deployments 3. the deployment stores that pull step in Prefect 4. later, when a worker starts a run for that deployment, the worker executes: - 
set_working_directory(directory="/app")
5. then the flow entrypoint is loaded from that directory ## Where the code runs The flow run code runs in the infrastructure started by the worker / work pool job. So 
/app
must exist in the run environment, not just where you executed 
prefect deploy
. If your worker container has:
Copy code
volumes:
  - ./:/app
working_dir: /app
then this can work if the actual flow run also executes in that same filesystem context. --- ## How to replicate this in 
deploy.py
The closest Python equivalent is usually:
Copy code
flow.from_source(
    source="/app",
    entrypoint="flows.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
)
That is the Python-native way to say: - the source code is in 
/app
- the flow entrypoint is relative to 
/app
For local storage sources, Prefect generates the equivalent pull behavior for you. ## Example
Copy code
from prefect import flow

@flow
def my_flow():
    print("hello")

if __name__ == "__main__":
    my_flow.from_source(
        source="/app",
        entrypoint="flows.py:my_flow",
    ).deploy(
        name="my-deployment",
        work_pool_name="my-pool",
    )
This is the best match to your YAML pattern. --- ## Key mental model ### In YAML
Copy code
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /app
entrypoint: flows.py:my_flow
means:
“At run time, switch into 
/app
, then load 
flows.py:my_flow
.”
### In Python
Copy code
my_flow.from_source(
    source="/app",
    entrypoint="flows.py:my_flow",
).deploy(...)
means nearly the same thing. --- ## Is there a direct 
pull=[...]
Python equivalent? Not in the simple deployment API pattern you’re likely using. In practice, if you want the YAML 
pull
behavior in Python, use: - 
from_source(...)
for source-based deployments, or - 
image=...
for image-based deployments Trying to manually re-express 
set_working_directory
as a raw pull step is generally not the preferred Python path. --- ## Which should you use? ### Use 
from_source("/app", ...)
if: - your worker/run environment has the repo mounted at 
/app
- you want behavior similar to your current YAML ### Use an image if: - you want portable deployments - you don’t want to rely on host bind mounts - you want the worker/run environment to always have the same code layout --- ## Important caveat If 
prefect deploy.py
is run somewhere that does not have 
/app
, that may still be a problem depending on how you construct the deployment. So if your deploy script runs inside the same container where 
/app
exists, this is fine. --- Docs: - prefect.yaml guide - Deploy via Python - Store flow code If you want, I can convert one of your exact YAML deployments into a 
deploy.py
version line by line.
l
@Marvin /app exists inside the container. Show the complete example with python deploy.py. My prefect-worker container looks like
Copy code
prefect-worker:
    image: prefecthq/prefect:3.6.29-python3.13
    container_name: prefect-worker
    command: >
      bash -c "pip install uv &&
               uv pip install --system -e /app &&
               prefect worker start --pool default-agent-pool"
    environment:
      PREFECT_API_URL: <http://prefect-server:4200/api>
    volumes:
      - ./:/app
    working_dir: /app
    depends_on:
      timescaledb:
        condition: service_healthy
      prefect-server:
        condition: service_started
    restart: unless-stopped
    env_file:
      - .env
m
✅ thought for 16.9 seconds
Yes — given that setup, 
/app
exists inside the worker container, so a Python 
deploy.py
using 
from_source(source="/app", ...)
is the closest equivalent to your YAML 
pull
step. ## What this will do Your YAML:
Copy code
pull:
  - prefect.deployments.steps.set_working_directory:
      directory: /app
roughly becomes this in Python:
Copy code
my_flow.from_source(
    source="/app",
    entrypoint="flows.py:my_flow",
).deploy(...)
That tells Prefect: - the code source is 
/app
- at run time, load the flow from the given entrypoint under 
/app
--- # Complete example ## 1. 
flows.py
Copy code
from prefect import flow, task, get_run_logger

@task
def print_resource(resource: str):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Processing resource: {resource}")

@flow
def sync_data(resource: str, limit: int = 100):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Starting sync for resource={resource}, limit={limit}")
    print_resource(resource)
    <http://logger.info|logger.info>("Done")
--- ## 2. 
deploy.py
This creates multiple deployments from the same flow, using different parameters.
Copy code
from flows import sync_data

DEPLOYMENTS = [
    {
        "name": "sync-customers",
        "parameters": {
            "resource": "customers",
            "limit": 100,
        },
    },
    {
        "name": "sync-orders",
        "parameters": {
            "resource": "orders",
            "limit": 200,
        },
    },
    {
        "name": "sync-products",
        "parameters": {
            "resource": "products",
            "limit": 50,
        },
    },
]

if __name__ == "__main__":
    for deployment in DEPLOYMENTS:
        sync_data.from_source(
            source="/app",
            entrypoint="flows.py:sync_data",
        ).deploy(
            name=deployment["name"],
            work_pool_name="default-agent-pool",
            parameters=deployment["parameters"],
        )
--- ## 3. How to run it Run 
deploy.py
from inside an environment where 
/app
exists and contains your project. Since your worker container already has that mount, you can run:
Copy code
docker exec -it prefect-worker python /app/deploy.py
That will register the deployments in Prefect. --- # Why this matches your worker setup Your worker container:
Copy code
image: prefecthq/prefect:3.6.29-python3.13
volumes:
  - ./:/app
working_dir: /app
means: - your repo is mounted into the container at 
/app
- Prefect worker starts from 
/app
- your flow source path 
/app
is valid inside that container So when the deployment runs, Prefect can resolve:
Copy code
source="/app"
entrypoint="flows.py:sync_data"
correctly. --- # If you want schedules too You can add a schedule directly in Python. Example with an interval schedule:
Copy code
from datetime import timedelta
from flows import sync_data

DEPLOYMENTS = [
    {
        "name": "sync-customers",
        "parameters": {"resource": "customers", "limit": 100},
    },
    {
        "name": "sync-orders",
        "parameters": {"resource": "orders", "limit": 200},
    },
]

if __name__ == "__main__":
    for deployment in DEPLOYMENTS:
        sync_data.from_source(
            source="/app",
            entrypoint="flows.py:sync_data",
        ).deploy(
            name=deployment["name"],
            work_pool_name="default-agent-pool",
            parameters=deployment["parameters"],
            interval=timedelta(hours=1),
        )
If you need cron instead, that can also be done in Python. --- # If you want triggers too If you have lots of repeated triggers, Python is especially helpful. For example: ``` from flows import sync_data def upstream_completion_trigger(flow_name: str, deployment_name: str): return { "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_name, }, { "prefect.resource.role": "deployment", "prefect.resource.name": deployment_name, }, ], } if name == "__main__": sync_data.from_source( source="/app", entrypoint="flows.py:sync_data", ).deploy( name="sync-downstream", work_pool_name="default-agent-pool", parameters={"resource": "downstream", "limit": 10}, triggers=[ { "type": "compound", "require": "any", "triggers": [ upstream_completion_trigger("flow_a", "deployment_a"), upstream_completion_trigger("flow_b", "deployment_b"), ], } ], )
Copy code
This is much more maintainable than large repeated YAML blocks.

---

# Important caveat

This pattern works well *because your worker runtime also has `/app` mounted*.

If later you switch to:
- Kubernetes workers
- ECS
- Docker workers launching separate containers
- remote execution environments

then `/app` may not exist there unless you deliberately mount or bake the code in.

In those cases, a more portable pattern is:
- build an image containing the code, or
- use Git/remote source

But for your current Docker Compose worker setup, `/app` is fine.

---

# Recommended structure for your setup

I’d suggest:
/app flows.py deploy.py pyproject.toml
Copy code
Then:
1. worker starts
2. `uv pip install --system -e /app`
3. run `python /app/deploy.py`
4. worker executes runs from `/app`

---

# Summary

For your container setup, the Python equivalent of:
pull: - prefect.deployments.steps.set_working_directory: directory: /app
Copy code
is effectively:
flow.from_source(source="/app", entrypoint="flows.py:my_flow").deploy(...)
Copy code
and a complete minimal `deploy.py` is:
from flows import sync_data DEPLOYMENTS = [ {"name": "sync-customers", "parameters": {"resource": "customers", "limit": 100}}, {"name": "sync-orders", "parameters": {"resource": "orders", "limit": 200}}, ] if name == "__main__": for deployment in DEPLOYMENTS: sync_data.from_source( source="/app", entrypoint="flows.py:sync_data", ).deploy( name=deployment["name"], work_pool_name="default-agent-pool", parameters=deployment["parameters"], ) ``` Docs: - Deploy via Python - Store flow code If you want, I can next show a version of 
deploy.py
that includes: - shared schedule - shared concurrency settings - generated triggers - and “latest only” self-skip logic patterns.
3 Views
Open in Slack
PreviousNext
Powered by