Prefect Cheat Sheet

1Install & first flow@flow

pip install prefect★
Python 3.9+. Bundles the CLI, an embedded SQLite-backed server, and the UI. (uv add prefect works too.)
from prefect import flow @flow def pipeline(name: str): print(f"hello {name}") if __name__ == "__main__": pipeline("world")★
A flow is any function wrapped in @flow. Call it like normal Python — Prefect creates a tracked flow run with logs, timing, and state.
@flow(name="etl", log_prints=True, retries=2) def etl(): ...★
Flow options: name, retries/retry_delay_seconds, log_prints (capture print), timeout_seconds, description. etl.with_options(retries=5) clones with overrides.
prefect server start # local UI at http://127.0.0.1:4200
Run the self-hosted server + UI to see runs (3.8 ships the redesigned UI by default). Or connect to Prefect Cloud (card 15).

2Tasks & futures@task

from prefect import flow, task @task def extract(url): return requests.get(url).json() @flow def etl(): data = extract("https://api/x")★
Tasks are the retriable, cacheable units. Call them from inside a flow — each call is a tracked task run.
fut = extract.submit(url) # run concurrently, returns a future result = fut.result()★
.submit() runs a task in the flow's task runner (concurrent by default) and returns a future; .result() blocks for the value.
async def etl(): ... # @flow / @task support async natively
Async flows/tasks work out of the box — await your tasks for I/O-bound concurrency.
# calling a @task outside a @flow just runs the functionnote
Tasks orchestrate only within a flow. Outside one they execute as plain functions (no tracking/retries).

3Dependencies & mappingfan-out & order

results = process.map([a, b, c]) # one task run per item★
.map() spawns a task run per item — parallel fan-out with no loop. Returns a list of futures; pass to another .map() to chain fan-outs.
from prefect import unmapped process.map(items, config=unmapped(cfg))★
unmapped(x) broadcasts one static value to every mapped run instead of iterating it — the common way to pass shared config into a .map().
notify.submit(wait_for=[load_future]) # ordering without data flow
wait_for= forces a task to wait on upstreams it doesn't take as arguments — express control dependencies explicitly.
from prefect import allow_failure cleanup(allow_failure(may_fail_future))
allow_failure() lets a failed upstream pass downstream without raising — use only when you deliberately handle the failed state.

4Retries, caching & loggingresilience

@task(retries=3, retry_delay_seconds=[1, 10, 60])★
Automatic retries with fixed or backoff delays (exponential_backoff(backoff_factor=2)). retry_condition_fn to retry only on certain failures; retry_jitter_factor to spread them.
from prefect.cache_policies import INPUTS, TASK_SOURCE, RUN_ID @task(cache_policy=INPUTS + TASK_SOURCE, cache_expiration=timedelta(hours=1))★
Caching skips re-running a task when its key is unchanged. Compose policies with + (INPUTS, TASK_SOURCE, RUN_ID, FLOW_PARAMETERS) or write a cache_key_fn. NO_CACHE to disable.
from prefect import get_run_logger logger = get_run_logger(); logger.info("processed %d rows", n)
Use the run logger so messages show in the UI attached to the run (or set log_prints=True on the flow). 3.8 makes URLs in logs clickable.
@task(tags=["db"], timeout_seconds=30, persist_result=True)
Tags drive concurrency limits (card 7); timeouts fail long-hanging tasks; persist_result saves outputs for reuse.

5States & resultsobservability

# states: Scheduled -> Pending -> Running -> Completed / Failed / Crashed★
Runs move through states you see in the UI. Retrying, Cancelled, Paused, and Crashed (infra died) are states too.
state = etl(return_state=True) if state.is_failed(): ... state.result()★
return_state=True gives the State object instead of the return value — inspect success/failure and pull the result.
@task(persist_result=True, result_storage="s3-bucket/prefect")
Results can be persisted to storage (S3/GCS/local) so downstream runs & caching can reuse them; tune result_serializer (pickle/json).
from prefect.states import Completed, Failed return Completed(message="skipped: no new data")
Return an explicit State to control run outcome — e.g. mark a no-op as Completed, or fail with a custom message.

6Transactionsatomic & idempotent

from prefect.transactions import transaction with transaction(): write_staging(); promote() # rolls back on failure★v3
Prefect 3 transactions group steps so a failure triggers rollback hooks — the foundation for atomic, idempotent pipelines.
@write_file.on_rollback def cleanup(txn): os.remove(txn.get("path"))★
Register compensating actions per task with @task.on_rollback (and .on_commit). If a later step in the transaction fails, Prefect runs the rollbacks in reverse.
with transaction(key="nightly-load"): ... # idempotency key
A transaction key makes the block idempotent — committed work isn't repeated on re-run. Transactions also gate result persistence (all-or-nothing).

7Concurrency & rate limitsparallel & protect

from prefect.task_runners import ThreadPoolTaskRunner @flow(task_runner=ThreadPoolTaskRunner(max_workers=8))★
Task runners drive concurrency. Threaded by default; DaskTaskRunner/RayTaskRunner (extra packages) distribute across a cluster. Subflows: call a flow inside a flow for a nested tracked run.
from prefect.concurrency.sync import concurrency with concurrency("database", occupy=1): run_query()★
Global concurrency limits: the concurrency() context manager acquires/releases slots from a named limit — cap access to a shared resource across every flow/worker. Async variant in prefect.concurrency.asyncio.
from prefect.concurrency.sync import rate_limit rate_limit("external-api") # throttle calls/sec
Rate limits throttle how frequently an operation runs (requests/second) — stay under third-party API quotas. Create limits with prefect gcl create.
prefect concurrency-limit create db 5 # tag-based task limit
The older tag-based limit caps concurrent task runs carrying a tag (e.g. at most 5 tasks tagged db). Complements the slot-based concurrency() above.

8Assets & @materializedata lineage

from prefect.assets import materialize @materialize("s3://data/customers.parquet") def build_customers(): ...★assets
Assets model the data your workflows produce (a table, file, model). @materialize turns a task into one that produces an asset (keyed by URI) — Prefect builds a lineage graph in the UI. The evolution of "from @task to @materialize."
@materialize("s3://data/report.csv", asset_deps=["s3://data/customers.parquet"])★
asset_deps declares upstream assets this one depends on — edges in the lineage graph even across flows/deployments.
from prefect.assets import add_asset_metadata add_asset_metadata("s3://data/report.csv", {"rows": n})
Attach runtime metadata (row counts, schema, freshness) to a materialization — visible on the asset in the UI. At least one asset must be given at decoration time.

9serve() & schedulesthe easy path

etl.serve(name="nightly", cron="0 2 * * *")★
flow.serve() creates a deployment and starts a long-running process that executes scheduled/triggered runs — the fastest way to productionize on one machine.
from prefect import serve serve(etl.to_deployment("a", cron="@daily"), report.to_deployment("b", interval=3600))
Serve several deployments from one process. Schedules: cron, interval (seconds), or RRule; add timezone=.
etl.serve(..., parameters={"env": "prod"})
Bake default parameters; override per run from the UI/CLI/API. Parameters are validated (Pydantic).

10Deploymentsremote runs

etl.deploy(name="prod", work_pool_name="k8s", image="myrepo/etl:latest", cron="@hourly")★
A deployment is the server-side record of how/when/where a flow runs. .deploy() builds & pushes an image and registers it to a work pool.
flow.from_source("https://github.com/me/repo", entrypoint="etl.py:etl").deploy(...)★
from_source() pulls flow code from Git/remote storage at run time — no image rebuild for code-only changes.
# prefect.yaml — declarative deployments (built with `prefect deploy`) deployments: - name: prod entrypoint: flows/etl.py:etl work_pool: {name: k8s} schedules: [{cron: "@hourly"}]
prefect.yaml is the version-controlled, multi-deployment config with build/push/pull steps. Deploy with prefect deploy; trigger with prefect deployment run 'etl/prod'.

13Automations & artifactsreact & report

# Automations (UI): on flow-run Failed -> send Slack / email★
Automations fire actions on triggers (run failed, deployment stuck, custom events, metric thresholds) — alerting, retries, cancellations, webhooks.
from prefect.artifacts import create_table_artifact, create_markdown_artifact create_markdown_artifact("## Report\n...", key="summary")★
Artifacts publish tables, markdown, links, images, and progress to a run for humans — surface results in the UI. Keyed artifacts show a history over time.
# SLAs (Cloud): alert when a deployment runs late / too long
Prefect Cloud adds SLAs & incident tooling on top of automations for on-call workflows.

14Events & triggersevent-driven

from prefect.events import emit_event emit_event("data.arrived", resource={"prefect.resource.id":"file/1"})★
Emit custom events from anywhere; Prefect records a stream you can build reactive automations on. Every object change also emits a lifecycle event (3.7+).
from prefect.events.schemas.deployment_triggers import DeploymentEventTrigger
Deployment triggers run a deployment when a matching event fires — e.g. "when data.arrived, run the load." Webhook-style, no polling.
etl.deploy(..., triggers=[{"enabled":True, "match":{"prefect.resource.id":"file/*"}}])
Attach triggers directly to a deployment (or in prefect.yaml) to make it event-driven.

15CLI, server & Cloudoperate it

prefect cloud login # or: prefect config set PREFECT_API_URL=...★
Point the CLI/SDK at Prefect Cloud (hosted, with RBAC, events, SLAs) or your self-hosted server.
prefect flow-run ls · prefect deployment ls · prefect work-pool ls
Inspect runs, deployments, pools from the terminal. prefect flow-run watch <id> streams a run live (3.7+).
prefect server start · prefect config view · prefect version
Run the local server + UI (redesigned by default in 3.8); view resolved settings/profiles. Re-run a flow with its options via the UI's run cloning.

16Ecosystem & integrationswhere it fits

# Prefect = Python-first, imperative; Airflow = DAG-centric; Dagster = asset-centric★
Prefect's pitch is minimal ceremony — add decorators to real code (now with assets when you want lineage). Airflow is the incumbent scheduler; Dagster is asset-first. All three orchestrate; pick by mental model.
pip install prefect-aws prefect-gcp prefect-dbt prefect-kubernetes★
Integration collections add blocks & tasks for clouds, warehouses, dbt, Kubernetes, Docker. prefect-dbt surfaces each dbt model with per-model observability (first-class dbt orchestration in Cloud, 2026).
# pairs with dbt, dlt, pandas/Polars, Dask/Ray for compute
Prefect orchestrates; it doesn't replace your transform/compute tools — it schedules & observes them.

Prefect cheat sheet

Outline

Flows & tasks

State & concurrency

Assets & deploying

Infra & config

Observe & ecosystem

Flows & Tasks

State & Concurrency

Assets & Deploying

Infra & Config

Observe & Ecosystem

Worth memorizing