Data Passing & Templating

35XCom (Python)#

What it does: The mechanism for passing small values between tasks. Inside a Python callable, ti.xcom_push(key, value) writes a value; ti.xcom_pull(...) reads one back.

Parameters (xcom_pull):

How it works:

• Every non-None return value from a callable is automatically pushed to the "return_value" key on success — you don't need to call xcom_push yourself just to pass your function's result downstream.
• Returning None (or nothing) from a callable pushes nothing — a downstream xcom_pull for that key returns None. Returning "" (empty string) does get pushed, and is distinct from None.
• Values are always JSON-encoded — pushing the integer 5 and the string "5" are stored, and round-trip, as distinct types.
• If a mapped task instance (map_index >= 0) calls xcom_pull(task_ids=X) with no explicit map_indexes, it defaults to pulling the matching index from X — pass map_indexes="all" explicitly if you want every slice's value as a list.
• do_xcom_push=False suppresses only the automatic return_value push — it never blocks an explicit ti.xcom_push(...) call inside your callable.

Example:

def extract(**context):
    ti = context["ti"]
    ti.xcom_push(key="row_count", value=42)
    return {"status": "ok"}   # auto-pushed to "return_value"

def consume(**context):
    ti = context["ti"]
    count = ti.xcom_pull(task_ids="extract", key="row_count")      # specific key
    upstream_return = ti.xcom_pull(task_ids="extract")              # default key="return_value"
    many = ti.xcom_pull(task_ids=["extract", "load"])               # list -> list of values
    print(count, upstream_return, many)

extract_task = PythonOperator(task_id="extract", python_callable=extract, provide_context=True)
consume_task = PythonOperator(task_id="consume", python_callable=consume, provide_context=True)
extract_task >> consume_task

36TaskFlow XCom refs#

What it does: The @task decorator lets you call one task-decorated function and pass its result directly into another — the dependency edge and the XCom pull are wired automatically, no manual xcom_pull/xcom_push needed.

How it works:

• Only works when you pass the upstream result as a top-level argument — an XComArg nested inside a list or dict you construct yourself (e.g. transform(data=[extract_a(), extract_b()])) isn't detected. Pass each upstream result as its own top-level argument instead.
• Only op_kwargs are populated by @task calls — positional arguments to your decorated function are re-bound to their parameter names internally.
• Resolved values are opportunistically JSON-decoded — if the upstream task returned a plain string that also happens to be valid JSON (e.g. "123"), your downstream function receives the decoded type (int 123), not the original string.
• Calling the same @task-decorated function more than once in a DAG auto-suffixes the task_id (_2, _3, ...) so each call gets a distinct task.
• TaskFlow-inferred edges and explicit >> edges coexist fine — you can add extra manual dependencies alongside the automatic ones.

Example:

from dag_parser.dynamic.dag_context import task

with DAG(dag_id="taskflow_demo", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    @task
    def extract():
        return [1, 2, 3]

    @task
    def transform(data):          # `data` will be the real list at execution time
        return [x * 2 for x in data]

    @task
    def load(data):
        print(f"loading {data}")

    # Calling these wires both the dependency edges AND the XCom pulls automatically
    raw = extract()
    doubled = transform(raw)
    load(doubled)

37Auto XCom from operators#

What it does: Several built-in operators automatically push a return_value XCom on success, without any Python code — useful for wiring a non-Python task straight into a downstream ti.xcom_pull or TaskFlow reference.

How it works, per operator:

• The auto-push only happens if the task actually produced a non-empty result — a Bash command with genuinely empty stdout, or a query returning zero rows, produces no XCom row at all (xcom_pull returns None, not "").
• SQL-family operators only capture the first row — a query returning multiple rows has everything after row 1 silently left out of return_value. Use a Python task if you need more than one row passed downstream.
• Bash's return_value is the raw stdout string, not JSON-parsed — if your script prints JSON, a downstream Python task needs to json.loads() it itself.
• An SSH task's auto-XCom mixes stdout and stderr together into one string — unlike BashOperator, which keeps them separate.

Example:

# SQL: first row auto-pushed as {"id": 1, "name": "acme"} etc.
get_customer = SnowflakeOperator(
    task_id="get_customer",
    connection_id="warehouse_snowflake",
    sql="SELECT id, name FROM customers WHERE id = 1",
)

# Bash: raw stdout auto-pushed as a plain string
count_files = BashOperator(task_id="count_files", bash_command="ls /data | wc -l")

def use_upstream(**context):
    ti = context["ti"]
    row = ti.xcom_pull(task_ids="get_customer")        # {"id": 1, "name": "acme"}
    stdout_text = ti.xcom_pull(task_ids="count_files")  # raw stdout string, e.g. "42\n"
    print(row, stdout_text)

use_task = PythonOperator(task_id="use_upstream", python_callable=use_upstream, provide_context=True)
[get_customer, count_files] >> use_task

38Go templating (non-Python operators)#

What it does: Bash/SQL/HTTP/SSH-family operators (everything that isn't a Python-family operator) have their params rendered through a lightweight templating pass before execution.

Available tokens:

How it works:

• .DS/.TS are computed in the DAG's own timezone (feature 3), not UTC — even though storage is UTC, a DAG with timezone="America/New_York" sees its local calendar date here.
• A typo in the token structure itself (e.g. misspelling .Params) fails the task loudly. A typo in a key inside .Params/.Var (e.g. .Params.typo) just renders as an empty string — these look similar but behave very differently, so check carefully.
• The three macros (ds_add, ds_format, ts_add) return the original, unmodified string on bad input rather than erroring — double check you're feeding ds_add/ds_format a YYYY-MM-DD string and ts_add a full timestamp.
• A param with no {{ anywhere in it skips rendering entirely, at no cost.

Example:

regional_extract = BashOperator(
    task_id="regional_extract",
    bash_command=(
        "python extract.py --date={{ .DS }} "
        "--window_start={{ ds_add .DS -7 }} "
        "--compact_date={{ ds_format .DS \"20060102\" }} "
        "--region={{ .Params.region }} "
        "--api_key={{ .Var.extract_api_key }}"
    ),
)

39Jinja2 templating (Python operators)#

What it does: Python-family operators (PythonOperator, ExternalPythonOperator, PythonVirtualenvOperator) render their op_args/op_kwargs/templates_dict with a full Jinja2 engine, giving access to the complete context.

Available context:

How it works:

• Only fields the parser tracks as "template fields" get Jinja treatment — for Python operators that's op_args/op_kwargs/templates_dict. Non-Python operators use Go templating (feature 38) for their entire params instead.
• var.value.x / var.json.x raise a clear error on a missing key — unlike the Go side's silent-empty behavior (feature 38), a typo'd variable name here fails the task loudly.
• var.json.x parses the variable's stored value as JSON — use var.value.x instead for a plain string.
• data_interval_start/data_interval_end are only real datetime objects for cron/descriptor-scheduled runs (feature 23) — referencing them on a manually triggered, timetable-scheduled, or dataset-triggered run fails loudly rather than silently returning None.

Example:

def report(**context):
    print(f"ds={context['ds']}, region={context['params'].get('region')}")
    print(f"api_key={context['var'].value.get('extract_api_key')}")

report_task = PythonOperator(
    task_id="report",
    python_callable=report,
    provide_context=True,
    op_kwargs={
        "note": "{{ ds }} run for {{ dag.dag_id }}, "
                "window={{ macros.ds_add(ds, -7) }} to {{ ds }}, "
                "threshold={{ var.json.threshold_config }}",
    },
)

40Variables & Connections#

What it does: Variables are a flat, global key-value config store, readable from both Go templates (feature 38) and Python Jinja (feature 39). Connections store reusable external credentials, referenced by name from operators that need them.

How it works:

• Every task gets every Variable, with no per-DAG or per-role scoping — treat Variables as workspace-wide configuration, not a place to isolate team-specific secrets.
• Encrypted Variables are decrypted before ever reaching a task — encryption only protects the value at rest and in the Admin UI's list view; a task that references the key still sees the plaintext.
• Connections are looked up by the executor directly via connection_id, not through templating — you can't read a connection's password with {{ .Var.some_conn_password }}; you pass connection_id to an operator that knows how to resolve it (Snowflake, SQL, S3, etc.).
• A connection_id that doesn't exist only fails when the task actually runs, not when the DAG is parsed — double-check the name if a task suddenly starts failing with a "connection lookup failed" error.
• Both Variables and Connections are created/managed through the Admin/Config UI or API — there's no DAG-file syntax to create one, only to reference one by key/connection_id.

Example:

# Reference a Variable from a Bash task (Go templating)
notify = BashOperator(
    task_id="notify",
    bash_command="curl -X POST {{ .Var.slack_webhook_url }} -d 'text=job done'",
)

# Reference a Variable from a Python task (Jinja/context)
def use_variable(**context):
    threshold = context["var"].value.get("threshold_config")
    print(f"threshold={threshold}")

use_var_task = PythonOperator(
    task_id="use_variable", python_callable=use_variable, provide_context=True,
)

# Reference a Connection by connection_id (created beforehand via the UI/API)
extract = SnowflakeOperator(
    task_id="extract",
    connection_id="snowflake_prod",   # must already exist
    sql="SELECT * FROM sales.orders WHERE order_date = '{{ .DS }}'",
)