New in version 2.0.

If you write most of your DAGs using plain Python code rather than Operators, then the TaskFlow API will make it much easier to author clean DAGs without extra boilerplate, all using the @task decorator.

TaskFlow takes care of moving inputs and outputs between your Tasks using XComs for you, as well as automatically calculating dependencies - when you call a TaskFlow function in your DAG file, rather than executing it, you will get an object representing the XCom for the result (an XComArg), that you can then use as inputs to downstream tasks or operators. For example:

from airflow.decorators import task
from airflow.operators.email import EmailOperator

def get_ip():
    return my_ip_service.get_main_ip()

def compose_email(external_ip):
    return {
        'subject':f'Server connected from {external_ip}',
        'body': f'Your server executing Airflow is connected from the external IP {external_ip}<br>'

email_info = compose_email(get_ip())


Here, there are three tasks - get_ip, compose_email, and send_email.

The first two are declared using TaskFlow, and automatically pass the return value of get_ip into compose_email, not only linking the XCom across, but automatically declaring that compose_email is downstream of get_ip.

send_email is a more traditional Operator, but even it can use the return value of compose_email to set its parameters, and again, automatically work out that it must be downstream of compose_email.

You can also use a plain value or variable to call a TaskFlow function - for example, this will work as you expect (but, of course, won’t run the code inside the task until the DAG is executed - the name value is persisted as a task parameter until that time):

def hello_name(name: str):
    print(f'Hello {name}!')

hello_name('Airflow users')

If you want to learn more about using TaskFlow, you should consult the TaskFlow tutorial.


When running your callable, Airflow will pass a set of keyword arguments that can be used in your function. This set of kwargs correspond exactly to the context variables you can use in your Jinja templates.

For this to work, you need to define **kwargs in your function header, or you can add directly the keyword arguments you would like to get such as ti=None to have the task instance passed.


To use logging from your task functions, simply import and use Python’s logging system:

logger = logging.getlogger("airflow.task")

Every logging line created this way will be recorded in the task log.


The TaskFlow API is new as of Airflow 2.0, and you are likely to encounter DAGs written for previous versions of Airflow that instead use PythonOperator to achieve similar goals, albeit with a lot more code.

More context around the addition and design of the TaskFlow API can be found as part of its Airflow Improvement Proposal AIP-31: “TaskFlow API” for clearer/simpler DAG definition

Was this entry helpful?