Great question — you can absolutely make deployments fully declarative and pipeline-driven in Prefect 3.x. Here are the patterns that work best, with pros/cons and how to handle cleanup. High-level options - Option A (most declarative): Use a

prefect.yaml

project and deploy in CI with

prefect deploy

. This gives you a version-controlled manifest of build/push steps and all deployments. - Option B (keep your current Python approach): Maintain a single

deploy.py

that defines all flows/deployments and run it as a CI step. Treat that script as the source of truth. This is still declarative if you treat the code as config. How to integrate into a pipeline - Auth in CI (required for either approach) - Set these as CI environment variables:

PREFECT_API_URL=<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>>
    PREFECT_API_KEY=<service-account-or-user-api-key>

- Option A: Prefect project + CLI 1) Initialize once and commit the result:

prefect init --recipe docker

2) Define build/push steps and deployments in

prefect.yaml

, for example:

name: my-project

     build:
       - prefect_docker.deployments.steps.build_docker_image:
           id: build-image
           requires: prefect-docker>=0.3.1
           image_name: <http://registry.io/my-image|registry.io/my-image>
           tag: "{{ env.GIT_SHA }}"
           dockerfile: auto

     push:
       - prefect_docker.deployments.steps.push_docker_image:
           requires: prefect-docker>=0.3.1
           image_name: "{{ build-image.image_name }}"
           tag: "{{ build-image.tag }}"

     deployments:
       - name: etl-hourly
         entrypoint: flows/etl.py:etl_flow
         work_pool:
           name: my-pool
           job_variables:
             image: "{{ build-image.image }}"
         schedules:
           - cron: "0 * * * *"

3) In CI:

prefect deploy --all

- Option B: Python-based deployments (your current approach), but run via CI - Keep a

deploy.py

and run it in CI after building/pushing your image. Example:

from prefect import flow, deploy
    from prefect.flows import Flow

    # Optionally pull flows from source repos
    # f1 = Flow.from_source("<https://github.com/org/repo.git>", "flows/etl.py:etl_flow")
    # f2 = Flow.from_source("<https://github.com/org/repo.git>", "flows/report.py:report_flow")

    # Or import flows locally
    @flow
    def etl_flow(): ...
    @flow
    def report_flow(): ...

    if __name__ == "__main__":
        deployments = [
            etl_flow.to_deployment(
                name="etl-hourly",
                cron="0 * * * *",
                tags=["prod"]
            ),
            report_flow.to_deployment(
                name="report-daily",
                cron="0 2 * * *",
                tags=["prod"]
            ),
        ]
        # Use a prebuilt image from your CI
        deploy(
            *deployments,
            work_pool_name="my-pool",
            image="<http://registry.io/my-image:${GIT_SHA}|registry.io/my-image:${GIT_SHA}>",
            build=False,
            push=False,
        )

- This is idempotent: the same name

<flow_name>/<deployment_name>

will be updated if it exists, or created if not. Cleanup and reconciliation - Updates are idempotent - Re-running deploy will update parameters, schedules (replaced), triggers (replaced), job variables, tags, description, etc., for the same

{flow_name}/{deployment_name}

. - What’s not automatically cleaned up - If you remove a deployment from your manifest/script, Prefect will not auto-delete the old one. You should run a reconcile step to delete deployments that are no longer declared. - Simple reconcile pattern (Python): ``` import asyncio from prefect import get_client # Your desired set of deployment full names (flow/deployment) DESIRED = { "etl_flow/etl-hourly", "report_flow/report-daily", } async def reconcile():

async with get_client() as client: existing = await client.read_deployments() for d in existing: full_name = f"{d.flow_name}/{d.name}" if full_name not in DESIRED: await client.delete_deployment(d.id) if name == "__main__": asyncio.run(reconcile())

- CLI tools that help in scripts:
  - List deployments: `prefect deployment ls`
  - Delete one or all: `prefect deployment delete <FLOW>/<DEPLOYMENT>` or `prefect deployment delete --all`
- Work pools and blocks
  - Pools: create/update with `prefect work-pool create` and treat pool name as a constant in your deployments.
  - Blocks: if you create blocks (e.g., notifications), manage them in code or IaC and delete via the API if needed. There isn’t a first-class “reconcile blocks” command; use the API client to list/delete.
  - For full IaC, you can manage Prefect resources with Terraform or Pulumi. See <https://docs.prefect.io/v3/advanced/infrastructure-as-code|Infrastructure as Code>.

Can a Prefect flow/deployment do the release?
- Yes. You can have a “bootstrap” flow that calls `deploy()` or `Flow.from_source(...).to_deployment(...).` It works, but consider:
  - Ensure PREFECT_API_URL/API_KEY are set in the runtime.
  - Avoid image builds from within the bootstrap deployment; pass a pre-built image and set `build=False, push=False`.
  - Avoid circular behavior (e.g., the bootstrap flow repeatedly deploying itself on every run).
  - Operationally, CI tends to be simpler and more predictable for deploying/updating deployments.

Recommended pipeline layout
- Build & push image in CI (Dockerfile in repo):

docker build -t registry.io/my-image:${GIT_SHA} . docker push registry.io/my-image:${GIT_SHA}

- Deploy
  - Option A:

prefect deploy --all

- Option B:

python deploy.py ``` - Optional: Reconcile step to delete old deployments no longer declared. Relevant docs and examples - Deploy via Python: Deploy via Python - Define deployments with YAML: prefect.yaml guide - CI/CD patterns: CI/CD guide - Deployments concepts: Deployments concepts - IaC: Infrastructure as Code Quick answers to your specific questions - “Best way to integrate into a pipeline?” Use

prefect.yaml

+

prefect deploy --all

in CI for the most declarative setup; or keep a single

deploy.py

and run it in CI as the source of truth. - “What cleanup needs to be done?” Orphaned deployments aren’t auto-removed; add a reconcile step that deletes deployments not present in your manifest/script. Schedules and triggers are replaced on update; no special cleanup needed there. Delete unused pools/blocks via API or IaC if you create them dynamically. - “Is it possible to do this as a Prefect flow/deployment?” Yes, a bootstrap flow can call

deploy()

using a prebuilt image and

build=False, push=False

. CI is generally preferred, but bootstrap flows are fine when you need self-provisioning behavior. If you share how you’re building images today and whether you’d prefer YAML or Python as the source of truth, I’m happy to sketch a minimal repo structure or CI config for your setup.

Yes — you can make your release flow block on the bootstrapper and fail if it fails. And you can link it as a subflow for nice UI nesting. Two solid patterns: Pattern A: Block with run_deployment (simple) -

run_deployment()

will wait for completion by default (

timeout=None

), but it will not automatically fail the parent — you need to check the child’s final state and raise if it failed. - Use

as_subflow=True

to make it appear as a child run in the UI.

from prefect import flow
from prefect.deployments import run_deployment
from prefect.states import raise_state_exception

@flow
def release_flow(image_tag: str):
    # Kick off bootstrapper and wait for it to finish
    child = run_deployment(
        name="bootstrapper/release",
        parameters={"image_tag": image_tag},
        as_subflow=True,   # nest in UI
        timeout=None,      # block until final state
        poll_interval=5,
        tags=["release"],
    )

    # Propagate failure to this flow if the child failed
    if child.state.is_failed():
        raise_state_exception(child.state)

    # Optional: verify completion (should be final with timeout=None)
    if not child.state.is_final():
        raise RuntimeError("Bootstrapper did not reach a final state.")

Pattern B: Schedule then explicitly wait (lets you log state transitions) - Start the child, then call

wait_for_flow_run()

which can log state transitions to the parent logs. - This still won’t stream full child logs, but you’ll see transitions like “Running -> Completed/Failed”.

from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import raise_state_exception

@flow
def release_flow(image_tag: str):
    # Start immediately but return control to parent
    child = run_deployment(
        name="bootstrapper/release",
        parameters={"image_tag": image_tag},
        as_subflow=True,
        timeout=0,          # don't wait here
    )

    # Now wait and log state transitions in the parent logs
    child = wait_for_flow_run(
        child.id,
        timeout=3600,       # 1h
        log_states=True,    # logs state transitions into parent
        poll_interval=5,
    )

    if child.state.is_failed():
        raise_state_exception(child.state)

About logs “pass-through” - There isn’t a built-in way to stream all child logs into the parent as if it were a direct child-call. Your options: - Use

as_subflow=True

so the UI nests the child run and you can drill into its logs. - After completion, fetch and re-emit child logs into the parent if you really need them inline:

from prefect import get_client, get_run_logger
    from prefect.client.schemas.filters import LogFilter

    @flow
    async def release_flow(image_tag: str):
        logger = get_run_logger()
        child = await run_deployment(
            name="bootstrapper/release",
            parameters={"image_tag": image_tag},
            as_subflow=True,
            timeout=None
        )

        # Fetch and relay logs (careful with volume)
        async with await get_client() as client:
            logs = await client.read_logs(log_filter=LogFilter(flow_run_id=child.id))
            for log in logs:
                <http://logger.info|logger.info>(f"[child] {log.message}")

        if child.state.is_failed():
            from prefect.states import raise_state_exception
            raise_state_exception(child.state)

- Note:

log_states=True

with

wait_for_flow_run

only logs state transitions, not full logs. Common gotchas that cause “parent succeeds while child fails” - Calling

run_deployment(..., timeout=0)

or a small timeout and not waiting/checking the final state. - Not raising an exception in the parent after detecting

child.state.is_failed()

. References - Deploy via Python: Deploy via Python

- API client (reading logs, etc.): API Client If you share whether your release flow is sync or async, I can tailor a minimal version that fits your exact setup.

Short answer: there’s no built-in “transaction” that makes multiple deployment changes atomic in Prefect 3.x. Each API call (create/update/delete, schedule updates) is applied individually. However, you can get safe, repeatable roll-forward/rollback behavior with a staged (blue/green) pattern and a bit of bookkeeping. Recommended patterns 1) Blue/green with paused “candidate” deployments (safe and simple) - Create new deployments with a new name or version suffix and set them paused (or with no schedules). - Run health checks (e.g.,

run_deployment()

and wait for completion). - If healthy, “commit” by enabling schedules for the new deployments and disabling schedules for the old ones. If unhealthy, delete the new ones (or leave them paused) and keep the old ones active. This gives you a clean rollback: pause new, resume old. It’s not one atomic call, but the switch is controlled and reversible. Hints: - In Python, use

Flow.to_deployment(..., paused=True)

and apply. Then re-apply with

paused=False

when promoting. - With CLI, prefer managing schedules per deployment: - List/inspect schedules:

prefect deployment schedule ls

- Pause/resume/clear:

prefect deployment schedule pause|resume|clear

- Docs: prefect.yaml guide 2) Update-in-place with quick rollback (use pinned images) - Keep deployment names stable, but pin the image tag in

job_variables.image

(or via

image=...

in

.deploy()

). - Before updating, record the current image tag and schedule config. - Apply the update (new image tag, etc.). If anything fails post-deploy, “rollback” by re-applying the previous image tag and schedule config. - This requires you to snapshot the previous deployment config first (parameters, schedules, triggers, job variables) so you can re-apply them on failure. 3) Batch “all-or-nothing” behavior (client-side) - Stage: create all new deployments paused. - Validate: run canaries on all. - Commit: enable schedules for all (and disable old) only if validation passed for all. Otherwise, delete the staged ones. - This provides an “apply all or none” effect without true atomicity. Code sketches - Blue/green stage/commit/rollback in Python (synchronous flow shown for brevity): ``` from prefect import flow, get_client from prefect.deployments import run_deployment from prefect.states import raise_state_exception NEW = [ # tuples of (flow, deployment_name, cron) ("repo@sha", "flows/etl.py:etl_flow", "etl-flow-abc123", "0 * * * *"), ("repo@sha", "flows/report.py:report_flow", "report-flow-abc123", "0 2 * * *"), ] OLD = ["etl-flow-prev", "report-flow-prev"] @flow def release(): created = [] try: # 1) Stage: create new deployments paused for source, entrypoint, name, cron in NEW: from prefect.flows import Flow f = Flow.from_source(source=source, entrypoint=entrypoint) d = f.to_deployment(name=name, cron=cron, paused=True, tags=["candidate"]) d.apply(work_pool_name="my-pool", image="registry.io/app:abc123") # build/push done earlier created.append(name) # 2) Validate: run a canary on each new deployment and wait for name in created: fr = run_deployment(name=f"{name}/{name}", timeout=1800, as_subflow=True) if fr.state.is_failed(): raise_state_exception(fr.state) # 3) Commit: enable new schedules, disable old schedules # Re-apply new deployments with paused=False or use schedule resume commands. for source, entrypoint, name, cron in NEW: f = Flow.from_source(source=source, entrypoint=entrypoint) f.to_deployment(name=name, cron=cron, paused=False).apply(work_pool_name="my-pool") # Disable old schedules (either clear schedules or re-apply them paused) # e.g.,

prefect deployment schedule clear

via CLI or re-apply with paused=True.

except Exception: # Rollback: remove newly created deployments or leave them paused import asyncio async def cleanup(): async with await get_client() as client: for name in created: try: # name here should be "<flow_name>/<deployment_name>" await client.delete_deployment_by_name(name) # or read and delete by id except Exception: pass asyncio.run(cleanup()) raise ``` Notes: - The exact “delete by name” call may require reading the deployment and then deleting by id; you can also use CLI

prefect deployment delete <FLOW>/<DEPLOYMENT>

. - If you prefer to keep candidates for inspection, skip deletion and just keep them paused. Rollback options at a glance - Blue/green: pause new + resume old schedules (fastest rollback). - In-place: re-apply previous image tag and schedule/params you snapshot before the update. - Cleanup: delete newly created candidate deployments on failure. Other tips - Use immutable, pinned image tags (e.g., the commit SHA) so rollback is just switching tags/schedules. - If you use

prefect.yaml

, keep both “old” and “new” deployment definitions in config with different names; a “promotion” step enables new and disables old. - If you must mutate existing deployments in place, always snapshot first, then mutate, and re-apply the snapshot on failure. Relevant commands and docs - List/delete deployments:

prefect deployment ls

,

prefect deployment delete <FLOW>/<DEPLOYMENT>

- Manage schedules:

prefect deployment schedule ls|pause|resume|clear

- Deploy via Python: Deploy via Python - API client for custom reconcile/cleanup: API Client - Prefect YAML (declarative): prefect.yaml guide If you share which approach you prefer (blue/green with separate names vs. in-place updates), I can sketch a minimal, concrete “release” script for your repo that includes a canary run and automatic rollback.