Great question. In Prefect 3.x there are a few solid patterns to run a flow “on complete” of another and to iterate a deployment with different params without adding more resources. Here are the go-tos, plus when to use each. Summary of patterns - Event-driven chaining (recommended): use Automations or Deployment triggers to run Deployment B when Deployment A completes. Decoupled, no blocking, scales well. - Controller flow: use

run_deployment()

from a flow to iterate a deployment over a param list. Can be blocking (sequential) or non-blocking (enqueue). - Concurrency controls: enforce sequential execution by setting a concurrency limit on the deployment or work queue so runs serialize automatically without extra infra. 1) Event-driven chaining with Automations (Run Deployment action) - Create an Automation that listens for

prefect.flow-run.Completed

from your upstream deployment and runs the downstream deployment. You can pass parameters via Jinja from the upstream run. Example:

from datetime import timedelta
from prefect.automations import Automation, EventTrigger, Posture, RunDeployment

automation = Automation(
    name="Run downstream after upstream",
    trigger=EventTrigger(
        expect={"prefect.flow-run.Completed"},
        match_related={"prefect.resource.name": "upstream-deployment-name"},
        posture=Posture.Reactive,
        threshold=1,
        within=timedelta(0),
    ),
    actions=[
        RunDeployment(
            source="selected",
            deployment_id="UUID-of-downstream-deployment",
            parameters={
                "customer_id": {
                    "template": "{{ flow_run.parameters['customer_id'] }}",
                    "__prefect_kind": "jinja"
                }
            },
        )
    ],
).create()

Useful when: You want a clean “on complete” trigger, don’t want to block a worker, and want a decoupled pipeline. Links: - Automations overview - API: RunDeployment action 2) Deployment triggers (shorthand for Automations) - Attach an event trigger directly to a deployment to run it when another deployment completes. Example:

from prefect import flow, serve
from prefect.events import DeploymentEventTrigger

@flow
def upstream(): ...

@flow
def downstream(customer_id: int): ...

if __name__ == "__main__":
    up_dep = upstream.to_deployment(name="upstream-deployment")
    down_dep = downstream.to_deployment(
        name="downstream-deployment",
        triggers=[
            DeploymentEventTrigger(
                expect={"prefect.flow-run.Completed"},
                match_related={"prefect.resource.name": "upstream-deployment"},
                parameters={
                    "customer_id": {
                        "template": "{{ flow_run.parameters['customer_id'] }}",
                        "__prefect_kind": "jinja"
                    }
                },
            )
        ],
    )
    serve(up_dep, down_dep)

Links: - API: DeploymentEventTrigger - API: Flow.to_deployment 3) Controller flow that iterates a deployment with different params - Programmatically loop over a list of parameters and call the same deployment for each item. You can block and wait for each to finish or just enqueue them all. Sequential (blocking) iteration: ``` from prefect import flow from prefect.deployments import run_deployment @flow def controller(customers: list[int]): results = [] for cid in customers: fr = run_deployment( name="process-customer/process-deployment", parameters={"customer_id": cid}, timeout=0 # see note below ) results.append(fr) return results

Important deadlock note:
- If you only have a single worker slot and you call `run_deployment` while waiting for completion, the child run may never start (parent is occupying the only slot). To avoid this:
  - Prefer `timeout=0` (non-blocking) so you enqueue downstream runs and return immediately, and
  - Enforce sequential processing using concurrency limits (below), or
  - Run the controller in a separate lightweight work pool if you must block on child completion.

Blocking vs non-blocking:
- Blocking: default waits; set `timeout` to a positive number to bound the wait.
- Non-blocking: `timeout=0` schedules and returns immediately.

Links:
- <https://docs.prefect.io/latest/api-ref/deployments/#prefect.deployments.run_deployment|API: run_deployment>

4) Serialize runs without adding resources (concurrency limits)
- Deployment-level concurrency: set to 1 to guarantee only one run of that deployment at a time.
- Work queue concurrency: set the queue’s concurrency to 1 to serialize across multiple deployments routed through the same queue.

Examples:
- When deploying in Python:

my_flow.deploy( name="process-deployment", work_pool_name="my-pool", concurrency_limit=1, # serialize runs for this deployment )

- Or create a work queue with a concurrency limit of 1 and target your deployment at that queue. All enqueued runs will execute one-by-one on the same resources.

Useful when: You want to enqueue multiple runs but ensure they process sequentially without more infra.

5) Simple parent/child subflows
- If both flows live in the same codebase and you don’t need deployments between them, call one flow from another directly. This runs “in-process” and is inherently sequential if you call it synchronously.

from prefect import flow @flow def child(x: int): ... @flow def parent(items: list[int]): for x in items: child(x=x) ``` Note: This pattern doesn’t use deployments and doesn’t queue work to a pool; it just runs wherever the parent runs. Putting it together for your use case - If you want downstream runs “on complete” of upstream: use an Automation or a Deployment trigger. - If you want to iterate a deployment over many params without spinning up more resources: - Choose a work queue or deployment concurrency limit of 1, and - Either: - Use a controller flow that calls

run_deployment(..., timeout=0)

in a loop (enqueue all), or - Schedule them via Automations (e.g., upstream emits multiple events that each trigger one downstream run with different params). - Avoid blocking

run_deployment

if the downstream runs share the same single worker slot as the controller. Questions to tailor this: - Are you on Prefect Cloud or OSS? - Do the upstream params/outputs need to be passed into the downstream run? - How many workers/slots do you have on the target work pool? If it’s 1, I’ll suggest a deadlock-safe layout for the controller/queues. More docs - Automations - API: run_deployment - API: Flow.to_deployment - Work pools and queues