Prefect cheat sheet

Flows, tasks, deployments, work pools, and blocks — Prefect 3 patterns for teams who want plain Python pipelines.

Orchestration & ToolsIntermediate6 sections

Flows and tasks

@flow(name="daily-orders", log_prints=True) def daily_orders(run_date: str): rows = extract(run_date) load(transform(rows)): A flow is a decorated Python function. Control flow is real Python — loops and conditionals need no special operator.
@task(retries=3, retry_delay_seconds=60): Retries at the task level with exponential options. The most common production decorator after @task itself.
@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1)): Result caching keyed on inputs. Reruns of an unchanged extract skip the work entirely.
my_task.submit(x): Concurrent execution returning a future. Call .result() to await it; without submit, tasks run sequentially.
my_task.map(items): Fan-out over a collection, creating one task run per element with independent retries.

my_flow.deploy(name="prod", work_pool_name="k8s-pool", cron="0 6 * * *"): The Prefect 3 way to register a scheduled deployment from Python — no YAML required.
prefect deploy --all: Deploys everything defined in prefect.yaml. Preferred when deployments live in version control alongside code.
prefect deployment run 'daily-orders/prod': Trigger an ad-hoc run of a deployment from the CLI, including parameter overrides.
parameters={"run_date": "2026-08-01"}: Parameters are typed by the flow signature and validated with Pydantic before the run starts.

prefect work-pool create --type kubernetes k8s-pool: Work pools describe where runs execute. Process, Docker, Kubernetes, and ECS are the common types.
prefect worker start --pool k8s-pool: Workers poll a pool and launch infrastructure per run. No worker means runs sit in Late state forever.
job_variables={"image": "myrepo/pipelines:1.4.0"}: Per-deployment infrastructure overrides — the clean way to pin an image without a new work pool.

SnowflakeConnector.load("prod-warehouse"): Blocks are named, reusable configuration stored server-side. The equivalent of an Airflow connection.
Secret.load("api-token").get(): Secret retrieval at run time. Never bake credentials into flow code or deployment parameters.
S3Bucket.load("lake").download_folder_to_path(...): Filesystem blocks double as storage backends for flow code and results.

get_run_logger(): Returns a logger whose output appears in the Prefect UI alongside the run. print statements need log_prints=True.
create_markdown_artifact(...): Attaches a rendered report to the flow run — row counts and data quality summaries where reviewers will see them.
@flow(on_failure=[notify_slack]): Hooks for failure and completion. Simpler than wiring callbacks on every task.
prefect flow-run logs <id>: CLI log retrieval when the UI is not open, useful in CI failure output.

def test_flow(): with prefect_test_harness(): assert daily_orders("2026-08-01") is not None: Runs the flow against a temporary local database so tests never touch the real API.
my_task.fn(x): Calls the undecorated function directly for pure unit tests without any orchestration overhead.

From DataLane — tutorials at/blog, practice SQL live in theplayground.