XCom, Connections, and Hooks

Tasks in a DAG are separate processes. Two problems follow: how does one task hand a value to the next, and how does a task get the address and password of the database it writes to without those being written into the code? Airflow's answers are XCom, Connections, and Hooks.

XCom: passing small values between tasks#

XCom ("cross-communication") lets one task publish a value that a later task reads. The value is stored in the metadata database, keyed by DAG run, task id, and a name.

Push and pull#

A PythonOperator callable that returns a value automatically pushes it:

def get_watermark(**context):    result = client.query("SELECT maxOrNull(created_at) FROM staging.orders")    return result.result_columns[0][0]     # pushed to XCom as "return_value"

read_watermark = PythonOperator(    task_id="read_watermark",    python_callable=get_watermark,)

A downstream task pulls it through the task instance (ti) in its context:

def load_since_watermark(**context):    watermark = context["ti"].xcom_pull(task_ids="read_watermark")    load_rows_after(watermark)

load_new = PythonOperator(    task_id="load_new",    python_callable=load_since_watermark,)
read_watermark >> load_new

xcom_pull(task_ids="read_watermark") reads that task's return_value for the current DAG run. You can also push explicitly with context["ti"].xcom_push(key="row_count", value=n) and pull with xcom_pull(task_ids="...", key="row_count").

XCom is for metadata, not data#

Every XCom value round-trips through the metadata database and is loaded into memory by the pulling task. It is sized for small values:

The pattern for large data is pass a pointer: task A writes the dataset to a table or to object storage and pushes its location; task B pulls the location and reads from there. The dataset never travels through XCom.

Connections: credentials out of code#

A Connection is a named record — stored by Airflow — that holds how to reach an external system: host, port, login, password, schema, and an extra JSON blob for anything else. It has a Conn Id like clickhouse_default or orders_postgres.

Without Connections, every task hardcodes this:

# in every callable, in every DAG — the problemclient = get_client(    host="clickhouse",    port=8123,    username="default",    password="password",)

The address and password are copied across files, committed to git, and have to be edited everywhere to change. A Connection centralises them.

Defining a Connection#

Three ways, same result:

# CLIairflow connections add clickhouse_default \  --conn-type generic \  --conn-host clickhouse --conn-port 8123 \  --conn-login default --conn-password password

# environment variable: AIRFLOW_CONN_<CONN_ID_UPPERCASE>, a URIexport AIRFLOW_CONN_CLICKHOUSE_DEFAULT='generic://default:password@clickhouse:8123'

Or the Admin -> Connections page in the web UI. In production the values come from a secrets backend (Vault, AWS Secrets Manager) that Airflow reads transparently.

Reading a Connection#

from airflow.hooks.base import BaseHook
def build_client():    conn = BaseHook.get_connection("clickhouse_default")    return get_client(        host=conn.host,        port=conn.port,        username=conn.login,        password=conn.password,    )

Now the DAG code has no credentials in it. Changing the password is one edit in one place.

Hooks: reusable clients over a Connection#

A Hook is a class that wraps a Connection and exposes methods to talk to that system. PostgresHook, HttpHook, S3Hook, and many provider hooks follow the same shape:

from airflow.providers.postgres.hooks.postgres import PostgresHook
def load_reference_data(**context):    hook = PostgresHook(postgres_conn_id="orders_postgres")    rows = hook.get_records("SELECT id, name FROM segments")    ...

The hook reads orders_postgres, opens the connection, and gives you get_records, get_pandas_df, run, and a raw connection if you need one — no host or password anywhere in the task.

If a system has no dedicated hook (ClickHouse, depending on your provider set), the pattern is the same by hand: read the Connection with BaseHook.get_connection, build your client from its fields, and keep that in one helper the tasks import.

Putting it together#

# common/clickhouse.py  — one helper, imported by every callablefrom airflow.hooks.base import BaseHookfrom clickhouse_connect import get_client
def clickhouse_client():    conn = BaseHook.get_connection("clickhouse_default")    return get_client(        host=conn.host, port=conn.port,        username=conn.login, password=conn.password,    )

# transform/raw_to_staging.pyfrom common.clickhouse import clickhouse_client
def load_raw_orders_to_staging(**context):    client = clickhouse_client()    since = context["ti"].xcom_pull(task_ids="read_watermark")    ...

Credentials live in the clickhouse_default Connection; the watermark travels as XCom; the DAG file just wires read_watermark >> load_raw_orders_to_staging.

Common mistakes#

Large payloads in XCom#

A DataFrame or result set in XCom bloats the metadata database and slows every run. Push a table name or file path; keep the data out of band.

Credentials in the DAG file#

Hardcoded hosts and passwords get committed and copied. Put them in a Connection and read it with a hook or BaseHook.get_connection.

One Connection per task, defined in code#

Defining connection details inline defeats the point. Define the Connection once (UI, CLI, env var, or secrets backend); reference it by conn_id.

Pulling XCom from the wrong task#

xcom_pull(task_ids=...) must name the task that pushed the value, and there must be a dependency edge so it has already run.

Assuming XCom crosses DAG runs#

XCom is scoped to a single DAG run. A value from yesterday's run is not visible today; for cross-run state, read it from a table.

Quick reference#

See also#