Airflow TaskFlow API: The Kwargs Typo That Broke Data Flow

A kwargs typo shipped, and downstream tasks received None for three days. Nothing raised. Nothing went red. The pipeline loaded empty values with a green Grid.

TaskFlow moves return values through XCom silently, so a wrong key reads as missing data, not an error. You'll learn why typed contracts beat silent plumbing.

We'll cover decorator wiring, multiple outputs, and mixing styles. Explicit outputs save pipelines.

Decorator Style vs Classic Operators

Decorator style writes pipelines as function calls. @dag marks the factory, @task marks the steps, and calling steps inside wires edges automatically. No bit-shifts, no manual XCom calls.

Classic style declares operators and links them with >>. It dominates legacy codebases and operator-heavy files. Both compile to identical graphs.

New pipelines default to decorators. Less boilerplate means fewer wiring typos, and typed signatures document the data contract inline.

How TaskFlow Wires Dependencies for You

Calling a @task function returns an XComArg representing its future value. Passing that arg into another task declares ordering and data flow together. The scheduler builds the edge; runtime fills the value.

Plain values also work as arguments and persist as task parameters. The mental shift is timing: DAG-file calls wire, worker runs execute. Nothing inside the task body runs at parse time.

Read the Graph view to confirm. Every function-call edge renders as a dependency arrow. Wiring you can see is wiring you can trust.

Behind the scenes it's still XCom: returns land under return_value (BaseXCom.XCOM_RETURN_KEY) and you can still xcom_pull(task_ids='transform') from classic code. You'll inspect values under the task's XCom tab and reuse logic with .override(task_id='start', retries=3) or imports from a shared module.

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["orders"],
)
def order_flow():
    @task
    def extract() -> dict:
        return {"orders": [101, 102, 103]}

    @task
    def transform(raw: dict) -> dict:
        return {"clean": raw["orders"]}

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

    load(transform(extract()))

order_flow()

Multiple Outputs: Unpacking and outputs Dicts

Multiple outputs travel as named dict entries. Return {"clean": rows, "rejected": bad} and let consumers unpack by key. Named access survives field additions; positional tuples do not.

Unpack at the call site: outputs = split() then outputs["clean"] into the next task. Each key becomes its own edge with its own lineage. The Graph view shows the fan-out honestly.

Document the dict as a schema. Keys, value shapes, and empty-case behavior belong in the docstring. Consumers code against the docstring, not the implementation.

Flag it explicitly: @task(multiple_outputs=True) splits each dict key into its own XCom you'd pull as order_summary["total_order_value"]. Without the flag the whole dict is one blob. You'll pass classic outputs via .output and chain mixes with >>.

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["orders"],
)
def order_multi():
    @task
    def split() -> dict:
        return {"clean": [101, 102], "rejected": [103]}

    @task
    def load_clean(rows: list) -> int:
        print(f"clean: {rows}")
        return len(rows)

    @task
    def quarantine(rows: list) -> int:
        print(f"rejected: {rows}")
        return len(rows)

    outputs = split()
    load_clean(outputs["clean"])
    quarantine(outputs["rejected"])

order_multi()

Partial Return Semantics

Partial returns mean some keys present, others None. A task that returns {"result": {...}} on success and None on skip creates a consumer lottery. Every downstream branch must handle absence.

Prefer explicit skip semantics over ambiguous Nones. Raise SkipMixin results or return a status envelope like {"status": "skipped"}. Consumers switch on status instead of guessing.

Test the partial shapes directly. Unit tests covering empty, partial, and full returns catch more than integration runs that only exercise the happy path.

Type Hints as Contracts

Type hints are contracts the editor can check. dict[str, list[int]] tells consumers exactly what flows. A misspelled key access lights up before commit, not after three days of loads.

Hints also guide XCom serialization. Builtins, dataclasses, and attr classes serialize cleanly. Exotic objects fail at runtime with confusing errors; hints steer authors toward serializable shapes.

Enforce with linting. A type checker in CI turns hint discipline from suggestion into gate. The incident's typo dies in the pull request.

Isolation matters at scale: @task.virtualenv(requirements=["colorama==0.4.0"]), @task.external_python(python="/path/to/python"), @task.docker(image="python:3.9-slim"), and @task.kubernetes(...) run the same function in different runtimes. You'll also meet @task.sensor(poke_interval=60, timeout=3600, mode="reschedule") returning PokeReturnValue and @task.run_if / @task.skip_if for runtime branching.

import pendulum
from airflow.sdk import dag, task

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["orders"],
)
def order_typed():
    @task
    def extract() -> dict[str, list[int]]:
        return {"orders": [101, 102]}

    @task
    def transform(raw: dict[str, list[int]]) -> dict[str, list[int]]:
        return {"clean": [o for o in raw["orders"] if o > 100]}

    @task
    def load(clean: dict[str, list[int]]) -> None:
        print(f"loading {clean['clean']}")

    load(transform(extract()))

order_typed()

Mixing TaskFlow and Classic Tasks in One DAG

Mixing works both ways. XComArgs feed classic operators as arguments, and TaskFlow tasks pull classic outputs by task_id. Migration proceeds file by file, not as a big bang.

Wrap legacy operators behind TaskFlow producers gradually. Keep one style dominant per file so reviewers read consistently. A file that is 90 percent TaskFlow with one legacy operator is honest; fifty-fifty is confusion.

Finish migrations. Mixed files are bridges, not destinations. Track remaining classic tasks and retire them on a schedule.

Templating still works: declare context you need (def f(*, ti, next_ds)) or grab it deep via get_current_context(), and set templates_exts=[".sql"] for file-loaded queries. You'll keep the DAG file wiring-only and unit-test the plain function with pytest before Airflow ever runs.