The Airflow TaskFlow API (@task Decorator)

Introduced in Airflow 2.0, the TaskFlow API lets you define tasks as plain Python functions decorated with @task, with dependencies inferred automatically from how you call those functions — rather than manually instantiating PythonOperator objects and wiring up XCom pulls and pushes by hand.

Classic Style vs TaskFlow Style

The same three-task pipeline, written both ways:

# Classic PythonOperator style
def extract(**context):
    data = {"order_count": 42}
    context["ti"].xcom_push(key="data", value=data)

def transform(**context):
    data = context["ti"].xcom_pull(key="data", task_ids="extract")
    return {"order_count": data["order_count"] * 2}

extract_task = PythonOperator(task_id="extract", python_callable=extract)
transform_task = PythonOperator(task_id="transform", python_callable=transform)
extract_task >> transform_task

# TaskFlow style — equivalent behavior, far less boilerplate
from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def orders_pipeline():
    @task
    def extract():
        return {"order_count": 42}

    @task
    def transform(data: dict):
        return {"order_count": data["order_count"] * 2}

    transform(extract())

orders_pipeline()

Notice what disappeared: no manual xcom_push/xcom_pull, no **context boilerplate, no separately-instantiated operator objects, and no explicit >> — calling transform(extract()) both passes data and establishes the dependency, because Airflow inspects the function call graph to build the DAG.

How XCom Passing Works Automatically

Every @task-decorated function’s return value is automatically pushed to XCom, and passing the result of one task call as an argument to another automatically creates both the dependency edge and the corresponding XCom pull — this single mechanism is what eliminates most of the classic style’s boilerplate. See XComs & Variables for what’s happening under the hood and its size limitations.

Mixing TaskFlow with Traditional Operators

TaskFlow doesn’t replace operators like BashOperator or provider operators — those still don’t have a decorator equivalent, and you’ll routinely mix both styles in the same DAG:

@dag(start_date=datetime(2026, 1, 1), schedule="@daily", catchup=False)
def mixed_pipeline():
    @task
    def get_config():
        return {"batch_size": 100}

    run_dbt = BashOperator(task_id="run_dbt", bash_command="dbt run")

    config = get_config()
    config >> run_dbt   # bitshift still works between TaskFlow tasks and classic operators

Dynamic Task Mapping

TaskFlow also enables dynamic task mapping — generating a variable number of parallel task instances at runtime based on data, something that was awkward to express with classic operators:

@task
def get_files():
    return ["a.csv", "b.csv", "c.csv"]   # the actual list is only known at runtime

@task
def process_file(filename: str):
    print(f"processing {filename}")

process_file.expand(filename=get_files())

.expand() creates one mapped task instance per item in the list returned by get_files(), each independently retryable and visible in the UI — a pattern classic operators can’t express nearly as cleanly.

Common Mistakes

Returning large objects from @task functions. Since return values go through XCom (backed by the metadata database by default), returning a large DataFrame or a big list will bloat the database — return references (file paths, table names) instead, the same rule that applies to classic XCom usage.

Forgetting to call the @dag-decorated function at the bottom of the file. orders_pipeline() at module level is what actually instantiates the DAG — omitting it means the file parses without error but no DAG appears in the UI.

Assuming TaskFlow eliminates the need to understand XCom. It hides the mechanics, but the same size limits and serialization requirements still apply underneath — debugging “why did this fail to serialize” still requires knowing XCom exists.

Frequently Asked Questions

Is TaskFlow required for new DAGs? No — classic operators remain fully supported and are still the right choice for provider operators and cases where explicit XCom key control is genuinely needed; TaskFlow is a productivity improvement, not a replacement requirement.

Can I use type hints with @task? Yes, and it’s good practice — Airflow doesn’t enforce them at runtime, but they make multi-task pipelines significantly easier to read and to catch mistakes in during code review.

Does .expand() have a limit on how many mapped tasks it can create? Yes — max_map_length (default 1024) caps this, both as a sane guardrail and because an unbounded number of mapped task instances would overwhelm the UI and metadata database.

Summary

The TaskFlow API turns Airflow’s often-verbose classic operator boilerplate into plain decorated Python functions, with dependencies and XCom data-passing both inferred automatically from how you call those functions. It doesn’t replace operators — you’ll still reach for BashOperator and provider operators regularly — but for pure-Python logic, it’s now the idiomatic default for new DAGs.