Apache Airflow Task SDK

1.2.0 · active · verified Thu Apr 09

The Apache Airflow Task SDK provides Python-native interfaces for defining Directed Acyclic Graphs (DAGs), executing tasks in isolated subprocesses, and interacting with Airflow resources at runtime. Its primary goal is to decouple DAG authoring from Airflow internals, offering a forward-compatible and stable interface across Airflow versions. The library is currently at version 1.2.0 and receives frequent updates, aligning with the release cadence of Apache Airflow 3.x.

Warnings

breaking Direct metadata database access from task code is now restricted in Airflow 3.x. Tasks can no longer directly import and use Airflow database sessions or models. All runtime interactions must go through a dedicated Task Execution API via the Task SDK.
Fix: Refactor task code to use the Task SDK's interfaces for interacting with Airflow resources (e.g., Connections, Variables, XComs) instead of direct database access.
breaking All DAGs must update their imports to refer to `airflow.sdk` instead of using internal Airflow modules directly. Deprecated legacy import paths (e.g., `airflow.models.dag.DAG`, `airflow.decorator.task`) will be removed in future Airflow versions.
Fix: Update all DAG and task-related imports to use the `airflow.sdk` namespace. For example, `from airflow.models.dag import DAG` becomes `from airflow.sdk import DAG`.
breaking SubDAGs are replaced by TaskGroups, Assets, and Data Aware Scheduling in Airflow 3.x. Existing SubDAGs will need to be refactored.
Fix: Migrate SubDAG implementations to use TaskGroups for grouping tasks, and explore Assets and Data Aware Scheduling for managing complex data dependencies.
deprecated SLAs (Service Level Agreements) are deprecated and replaced with Deadline Alerts in Airflow 3.x.
Fix: Transition from using SLAs to implementing Deadline Alerts for monitoring task and DAG adherence to time constraints.
gotcha The `catchup_by_default` DAG parameter is now `False` by default in Airflow 3.x. This can lead to unexpected behavior if DAGs were implicitly relying on catchup being enabled.
Fix: Explicitly set `catchup=True` in your DAG definition if you intend for it to run for past missed schedules. Otherwise, review your DAGs to ensure `catchup=False` is the desired behavior.
gotcha When passing context to tasks, the traditional `def my_task(**context)` signature is being superseded. While still functional, the recommended approach in the Task SDK is to explicitly retrieve context using `airflow.sdk.get_current_context()`.
Fix: Update task definitions to import `get_current_context` from `airflow.sdk` and call it within the task function to retrieve the execution context.

Install

pip install apache-airflow-task-sdk Install latest version

Imports

dag
```
from airflow.sdk import dag
```
task
```
from airflow.sdk import task
```
DAG
```
from airflow.sdk import DAG
```
Airflow 3.x and the Task SDK consolidate core authoring interfaces under `airflow.sdk`. Legacy imports will be removed.
BaseOperator
```
from airflow.sdk import BaseOperator
```
Use the stable `airflow.sdk` path for operators in Airflow 3.x.
Connection
```
from airflow.sdk import Connection
```
Variable
```
from airflow.sdk import Variable
```
Param
```
from airflow.sdk import Param
```
XComArg
```
from airflow.sdk import XComArg
```
XComArg is used for data passing between TaskFlow API tasks. TaskGroups replace SubDAGs in Airflow 3.x.
get_current_context
```
from airflow.sdk import get_current_context
```
The `get_current_context()` function is the recommended way to retrieve the execution context dictionary, offering a cleaner signature than `**context`.

Quickstart

This quickstart defines a simple Airflow DAG using the Task SDK's `@dag` and `@task` decorators. The `my_task` function is decorated as an Airflow task, and `example_simplest_dag` is a decorated function that defines the DAG structure and invokes the task. This pattern decouples DAG authoring from Airflow internals.

import pendulum
from airflow.sdk import dag, task

@task
def my_task():
    print("Hello from a Task SDK task!")

@dag(
    schedule=None,
    start_date=pendulum.datetime(2023, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
)
def example_simplest_dag():
    my_task()

example_dag_instance = example_simplest_dag()

view raw JSON →