Apache Airflow:
Basic concepts

The main trap when moving to 3.x

from airflow.models.taskinstance import TaskInstance

# Anti-pattern for 3.x: directly reading internal models from task code
def bad_task(**ctx):
    ti = TaskInstance.get_task_instance(
        dag_id="demo",
        task_id="step",
        run_id=ctx["run_id"]
    )
    return ti.state

Installing Airflow with Docker Compose

curl -LfO 'https://airflow.apache.org/docs/apache-airflow/stable/docker-compose.yaml'

mkdir -p ./dags ./logs ./plugins ./config

# AIRFLOW_UID — for correct permissions on mounted folders
# FERNET_KEY — encrypts sensitive fields in the metadata DB
echo -e "AIRFLOW_UID=$(id -u)\nFERNET_KEY=$(openssl rand -base64 32 | tr '+/' '-_')" > .env

docker compose up airflow-init   # one-time initialization
docker compose up

Why AIRFLOW_UID and FERNET_KEY are needed

export AIRFLOW_UID=$(id -u)
export AIRFLOW_GID=0
export FERNET_KEY=$(python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")

docker compose up airflow-init

• AIRFLOW_UID is needed for correct permissions on the volume.
• FERNET_KEY encrypts sensitive fields in the metadata DB.

Operator as a template: BashOperator

from airflow.operators.bash import BashOperator

say_hello = BashOperator(
    task_id="say_hello",
    bash_command="echo Hello",
)

DAG example

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

with DAG(
    dag_id="bash_dependencies_demo",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
) as dag:
    start = BashOperator(task_id="start", bash_command="date")
    task_a = BashOperator(task_id="task_a", bash_command="echo 'Processing A'; sleep 2")
    task_b = BashOperator(task_id="task_b", bash_command="echo 'Processing B'; sleep 3")
    finish = BashOperator(task_id="finish", bash_command="echo 'Done'")

    start >> [task_a, task_b] >> finish

Parallel and sequential execution

start >> [task_a, task_b] >> finish

start >> task_a >> task_b >> finish

Hooks

• Hook is Airflow’s software client to a service (Postgres, S3, GCS, Slack, etc.).
• Connection stores connection parameters: host, port, login, password, schema, and extras.

from airflow.providers.postgres.hooks.postgres import PostgresHook

def read_users():
    hook = PostgresHook(postgres_conn_id="analytics_pg")
    rows = hook.get_records("SELECT id, email FROM users LIMIT 100")
    return rows

Sensors

• Sensor is a special type of task that waits for a condition to be met before continuing.
• poke — periodic polling of a condition: the sensor checks every few seconds whether the event has occurred.
• deferrable — the task temporarily releases the worker and waits asynchronously via the Triggerer.
• deferrable is better for long waits because it saves worker slots.
• Why it is needed: sensors help run the next step only after a file, event, API response, or other data readiness condition appears.
• The choice depends on whether you need simple polling or more scalable asynchronous waiting.

from airflow.providers.standard.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id="wait_for_drop",
    filepath="/opt/airflow/data/input.csv",
    poke_interval=30,
    timeout=1800,
    deferrable=True,
)

Variables

• A Variable is a global key-value store in the metadata DB for runtime configuration of DAGs.
• Do not read a Variable at the top level of a DAG file; it slows down and complicates parsing.

from airflow.models import Variable

@task
def send_report():
    channel = Variable.get("alerts_channel", default_var="#data-alerts")
    return channel

XCom

@task
def extract():
    return {"cnt": 42}

ti.xcom_push(key="cnt", value=42)
ti.xcom_pull(task_ids="extract", key="cnt")

XCom in TaskFlow

• extract() returns a dictionary, and Airflow stores it as XCom.
• transform(data) gets the result as an argument, effectively doing an xcom_pull.
• As a result, one task passes data to the next without an explicit push/pull call.
• This is the convenient TaskFlow syntax for a chain of data processing between tasks.

@task
def extract():
    return {"raw_value": 42}

@task
def transform(data: dict):
    return data["raw_value"] * 2

transform(extract())