Flows, Tasks, and Subflows

Two decorators do almost everything in Prefect, which makes it easy to sprinkle them without asking what each one buys you. @flow and @task aren’t interchangeable garnish — they mark different units with different guarantees, and one of them, subflows, quietly fixes a problem Airflow spent years failing to solve well. This post is about using all three deliberately, and about the two dozen keyword arguments hiding behind those decorators that you’ll actually reach for once the toy example grows up.

What earns a @task

A @task is a unit Prefect will observe and, if you ask, retry, cache, and track as its own run. That’s the whole test: does this piece of work deserve to be seen and recovered independently? Calling a warehouse, hitting an API, reading a file — anything that can fail on its own, take real time, or that you’d want to see as a distinct row in the UI — earns a task. A three-line dictionary reshape between two of those does not; that’s a plain helper function, and wrapping it in @task only adds orchestration overhead and UI noise for something that never fails alone.

from prefect import flow, task

@task
def extract_orders() -> list[dict]:
    return [
        {"title": "Dune", "qty": 2, "price": 18},
        {"title": "Piranesi", "qty": 1, "price": 15},
    ]

def line_total(order: dict) -> int:      # plain helper, no decorator
    return order["qty"] * order["price"]

@task
def load(summary: dict) -> None:
    print(f"loaded {summary}")

@flow(log_prints=True)
def daily_load():
    orders = extract_orders()
    summary = {"count": len(orders), "revenue": sum(line_total(o) for o in orders)}
    load(summary)

extract_orders and load touch the outside world, so they’re tasks — each becomes a task run with its own state and logs. line_total is pure arithmetic that only runs inside a comprehension; leaving it a plain function keeps the UI honest about what actually did I/O. Coming from Airflow, resist the reflex that every step must be an operator. Here, “step the orchestrator cares about” and “function” are separate decisions, and most functions stay undecorated.

The full @task surface

The bare @task is the two-percent case. In a real pipeline the decorator carries the configuration Airflow spreads across operator kwargs, default_args, and the odd Variable. Here’s a task wearing most of what it can wear:

from datetime import timedelta
from prefect import task
from prefect.cache_policies import INPUTS

def alert_oncall(task, task_run, state):
    print(f"{task_run.name} failed: {state.message}")

@task(
    name="extract-orders",
    task_run_name="extract-{day}",
    tags=["warehouse", "extract"],
    retries=3,
    retry_delay_seconds=[10, 30, 60],
    retry_jitter_factor=0.2,
    timeout_seconds=120,
    cache_policy=INPUTS,
    cache_expiration=timedelta(hours=6),
    persist_result=True,
    log_prints=True,
    on_failure=[alert_oncall],
)
def extract_orders(day: str) -> list[dict]:
    print(f"pulling {day}")
    ...

Take these in groups, because they answer four different questions.

Identity and observability. name sets the task’s display name (it defaults to the function name); task_run_name templates the name of each individual run from the call’s arguments, so the UI shows extract-2026-05-05 instead of five identical extract_orders rows. That template pulls from the task’s parameters by name — {day} resolves to the day argument — which is the single highest-leverage thing you can do for a busy flow’s readability. tags attach labels that you can filter runs by and, more usefully, hang concurrency limits off of; a "warehouse" tag shared across every task that hits the warehouse is how you cap that pool globally, which the concurrency chapter picks up.

Failure and time. retries, retry_delay_seconds (a scalar or a per-attempt list), and retry_jitter_factor (spreads retry timing so a fleet of tasks doesn’t retry in lockstep and re-hammer a recovering service) are the recovery knobs. timeout_seconds bounds the run — overrun it and the task run goes to a failed state instead of hanging your flow forever, the thing Airflow’s execution_timeout does. These are Airflow-familiar in spirit; the retries chapter covers the sharper edges like retry_condition_fn.

Results and caching. cache_policy and cache_key_fn decide when a task run can be skipped because an equivalent one already succeeded — INPUTS here means “same arguments, same cached result.” persist_result forces the return value to durable storage (not just in-memory), and result_serializer picks how it’s written ("json", "pickle", or a custom serializer). This is the one area where the decorator argument is only the tip; caching in Prefect 3.x is a real subsystem and it gets its own results and caching chapter. Know that the knob lives here; learn the model there.

Logging and hooks. log_prints=True routes the task’s print calls into its run logs. on_completion and on_failure (each a list of callables) fire when the task run settles into that state — the local, in-code reaction that lives next to the task, as opposed to a platform-level automation. A hook receives (task, task_run, state), so it can read the message, inspect the result, page someone. The full state-hook story, including on_crashed, is in the states chapter.

You will almost never set all of these at once. The point is that the decorator is the configuration surface — there’s no separate operator class to subclass, no default_args dict to thread down — so when you need one of these behaviors you add a keyword and move on.

Task run versus flow run

The vocabulary lines up with Airflow’s, one rung shifted. A flow run is one execution of a flow — the counterpart of a DAG run. A task run is one execution of a task inside that flow run, the counterpart of a task instance. Trigger daily_load twice and you have two flow runs, each containing its own extract_orders and load task runs. States live at both levels: the flow run has a state, each task run has a state, and a task run failing is what typically drags its flow run to Failed.

The full @flow surface

@flow carries its own set, and it overlaps the task’s on purpose — a flow is also a run with a name, retries, and hooks — while adding the arguments that only make sense for the top-level unit.

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(
    name="daily-load",
    flow_run_name="daily-load-{day}",
    description="Extract, summarize, and reconcile a day of bookshop orders.",
    version="2026.05.05",
    task_runner=ThreadPoolTaskRunner(max_workers=8),
    retries=2,
    retry_delay_seconds=30,
    timeout_seconds=600,
    validate_parameters=True,
    log_prints=True,
    on_failure=[alert_oncall],
)
def daily_load(day: str):
    ...

flow_run_name templates the run’s name from parameters exactly like task_run_name. description defaults to the function’s docstring and shows in the UI; version tags the run with a version string (defaults to a hash of the source) so you can tell which revision of the code produced a given run — the manual cousin of Airflow 3’s DAG versioning. task_runner is the flow-only lever that decides how this flow’s tasks execute when submitted concurrently — the default runs them on a thread pool; swap in a Dask or Ray runner for distributed execution — and it’s the subject of its own discussion in the work pools chapter. retries, retry_delay_seconds, and timeout_seconds apply to the whole flow run: a flow with retries=2 re-executes the entire flow function on failure, which is a blunter instrument than task retries and worth using sparingly. State hooks (on_completion, on_failure, and flow-only additions) behave as they do on tasks. That leaves validate_parameters, which needs its own section.

Parameters and validation

A flow’s parameters are its interface — the values a schedule, an API call, or a UI form supplies when it triggers a run. Prefect leans on your type hints to validate and coerce those values through Pydantic v2 before the flow body runs. This is not decoration; it’s a gate.

from datetime import date
from prefect import flow

@flow
def daily_load(day: date, limit: int = 1000, dry_run: bool = False):
    print(type(day), day)   # <class 'datetime.date'> 2026-05-05
    ...

Trigger this from the UI or the API with day="2026-05-05" — a string, because JSON has no date type — and Prefect coerces it to a real datetime.date before your code sees it. Pass limit="1000" and it becomes an int. Pass day="not-a-date" and the flow run fails at validation with a clear Pydantic error before the body executes, instead of blowing up three lines in when you try to do date math. Coming from Airflow, this replaces the ritual of reaching into dag_run.conf and hand-checking every value; the type hint is the schema.

For anything beyond a scalar, hand the flow a Pydantic model and get structured validation for free:

from pydantic import BaseModel, Field
from prefect import flow

class LoadConfig(BaseModel):
    day: date
    limit: int = Field(gt=0, le=100_000)
    sources: list[str] = ["orders", "returns"]

@flow
def daily_load(config: LoadConfig):
    print(config.limit, config.sources)

Now the trigger sends a JSON object, Prefect parses it into a LoadConfig, and limit=0 or a negative limit is rejected by the Field constraint. The flow body receives a fully-typed, already-validated object. This is a genuine step up from conf dictionaries you defensively .get() your way through.

When you don’t want coercion — say a parameter is an exotic type Pydantic can’t build, or you’re passing an object you’d rather hand through untouched — set validate_parameters=False on the flow. Validation is skipped entirely and values arrive as given. It’s an escape hatch, not a default; leaving validation on is what makes a scheduled flow fail loudly at the boundary rather than quietly deep inside.

Return values, not XComs

Look at how summary moves from extract_orders to load: it’s a return value passed as an argument. That’s the entire mechanism. There is no xcom_push, no ti.xcom_pull, no separate metadata store that serializes your data behind the scenes and imposes a size limit. A task returns a Python object; you hand it to the next task the way you’d pass any value in any function.

In Airflow, XComs exist because tasks are separate operator instances that can’t share a call stack, so data between them has to detour through the metadata database — the whole reason the XComs post needs to exist. Prefect’s tasks run inside one Python function, so passing data is just passing data. Prefect does have a durable results layer underneath — where returns are stored so they survive retries and power caching — but you don’t reach for it to move a value between two steps. That’s covered later in the series; for now, a return is a return.

Getting state from a call

By default, calling a task returns its result — the return value — and if the task failed, the exception is raised right there in your flow, exactly like calling a normal function. Most of the time that’s what you want: let it raise, let the flow fail. But sometimes you need to inspect how a task ended without your flow immediately dying on it, and Prefect gives you a few precise handles.

return_state=True changes what the call hands back. Instead of the result (or a raised exception), you get the task run’s state object, and reading the value becomes an explicit, non-raising step:

@flow
def daily_load(day: date):
    state = extract_orders(day, return_state=True)
    if state.is_failed():
        print("extract failed; loading yesterday's snapshot instead")
        orders = load_snapshot(day)
    else:
        orders = state.result()   # unwrap the value now
    ...

state.result() pulls the return value out of a completed state; on a failed state it re-raises unless you pass raise_on_failure=False. This is the mechanism behind conditional recovery — you catch the failure as data, branch on it, and keep going, instead of the whole flow crashing. It’s also how a hook or an automation reads what a run produced.

allow_failure is the complementary tool for the dependency direction. Normally, passing one task’s output into another creates a data dependency: if the upstream failed, the downstream never runs, because there’s no value to give it. Wrap the argument in allow_failure and the downstream runs anyway, receiving the failed state instead of a value — the deliberate “run this even if its input broke” you’d express with trigger rules in Airflow:

from prefect import allow_failure

@flow
def daily_load(day: date):
    orders = extract_orders(day)
    report = build_report(orders)
    notify(allow_failure(report))   # send the notification even if the report failed

The states chapter goes deep on the full model — Failed versus Crashed, AwaitingRetry, the lot — so treat this as the working subset: return_state=True to inspect, .result() to unwrap, allow_failure to depend-through-failure. See the states and hooks chapter for the rest.

One more distinction lives here: calling a task directly versus .submit(). A direct call runs the task and blocks until it’s done — sequential, synchronous, the default. .submit() hands the task to the flow’s task runner and returns a PrefectFuture immediately, letting the flow keep going and the task run in the background:

@flow
def daily_load(days: list[date]):
    futures = [extract_orders.submit(d) for d in days]   # all start concurrently
    results = [f.result() for f in futures]              # gather when you need them

.submit() is how you get concurrency out of the task runner without touching threads yourself; .result() on the future blocks until that one finishes. Passing a future straight into another task’s arguments wires up the dependency and unwraps it automatically — you only call .result() when you need the value in flow code. Direct call for a simple sequence, .submit() when the steps are independent and you want them overlapping.

Async flows and async tasks

Because a flow is just a Python function, an async def flow is a first-class flow, and an async task is a first-class task. This matters the moment your pipeline is I/O-bound — a dozen HTTP calls, a handful of async database drivers — because you get real asyncio concurrency instead of thread-pool workarounds.

import asyncio
from prefect import flow, task

@task
async def fetch(source: str) -> dict:
    async with http_client() as c:
        resp = await c.get(f"https://feed.internal/{source}")
        return resp.json()

@flow
async def daily_load(sources: list[str]):
    results = await asyncio.gather(*(fetch(s) for s in sources))
    return combine(results)

You await an async task call the way you’d await any coroutine, and you can asyncio.gather a batch of them for concurrency without a task runner in sight. Run the flow with asyncio.run(daily_load([...])) or, from the deployment machinery later in the series, let Prefect await it for you. Sync and async tasks coexist in the same codebase; pick async where the work is genuinely awaitable and sync everywhere else — mixing them is fine, and Prefect doesn’t force the whole pipeline to commit to one color.

Subflows: a flow calling a flow

A flow can call another flow, and the callee becomes a subflow — a flow run nested inside the parent’s flow run, with its own state and its own group of task runs in the UI. That’s the composition primitive, and it’s the one Airflow never delivered cleanly. SubDAGs were fragile enough that they were removed; TaskGroups replaced them but are only visual grouping — they don’t give you a reusable, independently-stated unit you can call from anywhere.

Give the bookshop a reconciliation step and make it its own flow:

from prefect import flow, task

@task
def check_totals(summary: dict) -> bool:
    return summary["revenue"] > 0

@flow
def reconcile(summary: dict) -> None:
    if not check_totals(summary):
        raise ValueError("revenue reconciliation failed")

@flow(log_prints=True)
def daily_load():
    orders = extract_orders()
    summary = {"count": len(orders), "revenue": sum(line_total(o) for o in orders)}
    load(summary)
    reconcile(summary)

reconcile is a full flow — it has its own tasks and states and could be scheduled or tested on its own — and daily_load uses it by calling it. In the UI, the daily_load run shows a nested reconcile subflow run beneath it, its check_totals task run tucked inside. You compose pipelines the way you compose functions: extract a piece, give it a name, call it from wherever you need it. No SubDAG caveats, no accepting that your grouping is cosmetic.

Two properties are worth naming. First, a subflow runs in-process, in the same Python process and (unless you make it async and submit it) inline in the call — you pass it real Python objects and get real Python objects back, no serialization at the boundary. Second, state rolls up: a failed subflow, by default, fails its parent, the same way a failed task does. reconcile raising ValueError fails the reconcile subflow run, and that failure propagates up and fails daily_load — unless you called it with return_state=True and chose to handle it, exactly the mechanism from the previous section. Subflows are the level at which you assemble a big pipeline out of independently-testable pieces while keeping one coherent, drill-down-able run in the UI.

run_deployment: crossing the infra boundary

A subflow shares your process, which is its strength and its limit — everything runs on the same worker, at the same time, with the same dependencies. When you want one flow to trigger another deployment — a flow that lives elsewhere, runs on different infrastructure, or is owned by another team — the tool is run_deployment:

from prefect.deployments import run_deployment

@flow
def daily_load(day: date):
    summary = summarize(extract_orders(day))
    # hand off to a separately-deployed flow, on its own infra
    run_deployment(
        name="reporting/nightly-report",
        parameters={"day": day.isoformat(), "revenue": summary["revenue"]},
    )

This creates a run of the nightly-report deployment in the reporting project and (by default) waits for it, surfacing it as a related run. It’s the cross-infra composition primitive — the closest Prefect analog to Airflow’s cross-DAG triggering — and it’s how you stitch a fleet of independently-deployed flows into one lineage.

The critical difference from a subflow is the boundary. A subflow shares memory, so data passes as live objects. run_deployment crosses a process — often a machine — so the data you send serializes: parameters go out as JSON through the API and are validated by the target flow’s Pydantic signature on the way in, which is exactly why we send day.isoformat() and a plain number rather than rich objects. You get isolation, independent scaling, and separate ownership; you pay for it with a serialization boundary and the size and type limits that come with JSON. Reach for a subflow when the pieces belong to the same pipeline and process; reach for run_deployment when they’re genuinely separate systems that should stay decoupled. Deployments themselves are the next major chapter; this is just the seam where flows call across them.

Tags, naming, and nested tasks

Two habits pay for themselves the first time a flow run misbehaves at 2 a.m. Name your runs with flow_run_name/task_run_name templates so the UI reads as daily-load-2026-05-05 instead of a wall of identical function names — future-you scanning a list of runs will know instantly which day broke. Tag by concern, not by name: a "warehouse" tag on every task that touches the warehouse, a "external-api" tag on every task that leaves your network. Tags are how you filter runs across flows and, more importantly, where global concurrency limits attach — so the tag you add for observability today becomes the throttle you need next month.

A word on nesting tasks. Calling a task from inside another task is allowed, and in Prefect 3 the inner call does get its own tracked task run — you’ll see both in the UI and in the logs. (If you’ve read that nested tasks aren’t tracked, that was Prefect 2; it changed.) So the reason to avoid deep nesting isn’t lost observability any more — it’s that a task calling a task calling a task produces a run graph that reads like a stack trace rather than a pipeline, and you lose the flat, glanceable shape that made the flow legible in the first place. Nest when it genuinely helps; reach for a subflow when you want a named, reusable unit with its own state roll-up. The clean shape is flat — a flow orchestrates tasks; tasks call plain functions, not other tasks. When you genuinely need a nested unit of orchestration, that’s the signal to make the inner piece a subflow instead, which Prefect does track as a nested run with full state roll-up. Keep tasks as leaves and reach for subflows when you want a branch, and the run tree in the UI will mirror the structure you actually intended.

Final thoughts

The discipline Prefect asks for isn’t in an API — it’s in deciding what’s worth orchestrating, and then reaching for the right keyword. @task for the units that fail, wait, or deserve their own row; plain functions for everything in between; @flow for anything you’d want to name, version, validate, or call from another flow. The decorator arguments are the whole configuration surface, so the leap from a toy flow to a production one is a handful of keywords — retries, a task_run_name, a tag, a Pydantic-typed parameter — not a new class hierarchy. And when a pipeline outgrows one process, subflows compose in-memory and run_deployment composes across infrastructure, with the serialization boundary being the one real difference between them. Get those calls right and your pipeline reads as ordinary, composable Python that happens to be observed — which was the promise the whole time. Next we lean on that: when the workflow itself has to change shape with the data, Prefect’s answer is the least surprising thing an Airflow engineer could hear.

Next: Dynamic Workflows: Just Write Python