Airflow operators & sensors

Dexaflow runs any Apache Airflow operator or sensor — BigQueryInsertJobOperator, S3KeySensor, SnowflakeOperator, the ~1,500 operators across the provider ecosystem — without re-implementing a single one. It does this by keeping the pod-per-task execution model and instantiating the real Airflow class inside the task container, driven by a thin Go control plane and a small Python runtime.

This page documents how that integration works end to end, and every behaviour the runtime gives an operator: connection resolution, the run context ({{ ds }}, {{ var.value.X }}, params, data interval), XCom chaining between tasks, multi-key XCom, and the Open in … UI extra-links. The design rationale lives in ADR 0040; this page is the operator’s-eye view.

High-level architecture

The same Airflow SDK is touched at two points, for two different reasons:

• At compile time, a structural shim (ADR 0024) imports the user’s dag.py to capture the graph — without a scheduler, a database, or really running anything. It records each operator’s dotted class path and its constructor arguments into dag.json.
• At run time, inside the task pod, the real Airflow SDK is imported and the operator is actually executed — import_string(class)(**args).execute(context).

The Go control plane never imports Airflow. It speaks the Airflow-compatible /api/v2/ API and orchestrates pods; all Python lives at the two edges.

flowchart TB
  subgraph Dev["Author / CI"]
    DAG["dag.py<br/>(BigQueryInsertJobOperator, S3KeySensor, @task, bash…)"]
    SHIM["parser structural shim<br/>(_generic.py meta-path finder)<br/>captures __leoflow_operator_class__ + args"]
    DJ["dag.json<br/>type=airflow_operator<br/>operator_class + args"]
    DAG -->|dexaflow compile| SHIM --> DJ
  end

  DJ -->|dexaflow push| API

  subgraph CP["Control plane (Go) — no Airflow import"]
    API["HTTP API /api/v2 + /ui"]
    SCH["Scheduler"]
    STORE["agent_store<br/>resolves DagRun → TaskSpec"]
    API --- SCH --- STORE
  end

  STORE -->|"gRPC GetTaskSpec<br/>(agentrpc TaskSpec)"| AGENT

  subgraph POD["Task pod (DAG image)"]
    AGENT["dexaflow-agent (Go, PID 1)<br/>buildEnv: stamps DEXAFLOW_* / AIRFLOW_* env<br/>BuildCommand: picks runtime mode"]
    RT["leoflow_runtime (Python)<br/>run / run_operator / run_bash"]
    SDK["Apache Airflow SDK<br/>import_string(class)(**args).execute(context)"]
    AGENT -->|"argv + env"| RT --> SDK
  end

  SDK -.->|"return_value, _extra_links,<br/>custom XCom keys (files)"| AGENT
  AGENT -->|"gRPC PushXCom / SetExtraLinks"| STORE

The dispatch decision

dag.json carries a task type. The agent’s BuildCommand (internal/agent/command.go) maps it to a runtime invocation:

Only classes the shim actually captured carry the __leoflow_operator_class__ marker, so the compiler routes them to airflow_operator; everything else stays on its native fast path. This ordering matters — the class marker is checked before the Bash/Http/Python name-substring heuristic, so an operator whose name happens to contain “Http” is not mis-routed.

The agent ⇄ runtime contract

The Go agent and the Python runtime communicate entirely through environment variables and small files — there is no shared Airflow object. The agent’s buildEnv stamps the env; the runtime reads it. This is the integration’s spine.

Features

Generic operators & sensors

The runtime’s run_operator(class, args) does what Airflow’s scheduler would do for a synchronous task, minus the database:

cls = import_string(operator_class)        # the real provider class
op  = cls(task_id=…, **merged_args)         # constructed with the captured args
op.render_template_fields(context)          # Jinja on template_fields ({{ ds }} etc.)
op.execute(context)                         # the actual work

A template field that fails to render fails the attempt before execute(), as on Airflow: the task log names the field and carries the Jinja error, for example template field 'sql' of BigQueryInsertJobOperator failed to render: UndefinedError: 'nope' is undefined. The attempt then retries or fails like any other failure, and the task’s on_failure_callback runs on its last attempt. Before 0.5.5 the error was logged and the operator ran with the raw {{ ... }} text (ADR 0084).

Poke-mode sensors run on the same path: execute() drives the poke loop standalone; a timeout raises AirflowSensorTimeout, which surfaces as an ordinary task failure. The operator’s connection comes from AIRFLOW_CONN_<ID> — the same env-secrets mechanism the connectors work already delivers — so no Airflow metadata DB is involved.

Connections & variables

Connections and Variables are delivered as AIRFLOW_CONN_<ID> and AIRFLOW_VAR_<KEY> env vars and resolved by Airflow’s native env-secrets backend — exactly as they resolve for a hook inside a @task. In templates, {{ var.value.X }} and {{ conn.X }} resolve through accessors the runtime wires up in _secrets_accessors (preferring airflow.sdk.execution_time.context, with a fallback for older SDKs).

The full run context

_operator_context() builds the dict passed to execute() and to templating, so operators and native tasks see the same macros Airflow exposes:

ds, ds_nodash, ts, ts_nodash, ts_nodash_with_tz, logical_date, run_id, task_instance/ti, task_instance_key_str, try_number, task, dag, dag_run, data_interval_start, data_interval_end, params, macros, var, conn, and exception in an on_failure_callback.

logical_date is the run’s logical date as a timezone-aware datetime, or None on a run without one. ds_nodash, ts_nodash and ts_nodash_with_tz are derived from it in the Task SDK’s formats (20260614, 20260614T053000, 20260614T053000+0000) and, as on Airflow, are absent on a run without a logical date. macros is the Task SDK’s macros module ({{ macros.ds_add(ds, 1) }}, {{ macros.ds_format(...) }}, macros.datetime, macros.timedelta); in an image without the Task SDK the runtime supplies ds_add, ds_format and the same standard modules itself.

dag_run carries dag_id, run_id, logical_date, data_interval_start, data_interval_end, run_type (the Task SDK’s values: manual, scheduled, backfill, asset_triggered) and conf, the run’s conf alone. conf is not params: it holds what the run was triggered with, without the param defaults that params layers under it. On an operator run task is the operator itself; on a @task or bash task, and for dag everywhere, the runtime holds no Airflow object, so task and dag carry their ids ({{ task.task_id }}, {{ dag.dag_id }}) and nothing else.

The context objects carry only what the agent delivers. An attribute they lack, such as dag_run.start_date or ti.hostname, raises an AttributeError that names it and says Dexaflow does not provide it, instead of a bare error deep inside a provider. These Task SDK keys are not provided, because the task pod has no data for them: prev_data_interval_start_success, prev_data_interval_end_success, prev_start_date_success, prev_end_date_success (the previous successful run), templates_dict (not captured by the compiler), task_reschedule_count, the asset keys (inlets, outlets, inlet_events, outlet_events, triggering_asset_events, asset_state_store), task_state_store, expanded_ti_count, map_index_template, partition_key, partition_date, reason and test_mode. A template that names one renders it as an empty string (an empty quoted argument in a bash command). The runtime’s tests compare this list with the pinned Task SDK, so a Task SDK upgrade that adds a key fails until it is provided or listed (#1843).

params (#148) are layered as in Airflow: the DAG’s declared param defaults, then the task’s own params=, then the run’s conf, each winning per key. The control plane resolves them when the agent fetches the task spec and serializes them into DEXAFLOW_PARAMS, and the runtime merges the operator’s params= under them before rendering, so {{ params.x }} sees task params too (#1841, #1844). A BashOperator’s and a @task’s own params= are captured by the compiler as the task’s params in dag.json and layered the same way (#1843). The data interval, logical date, run conf and run type ride through the gRPC TaskSpec (proto fields logical_date, data_interval_start/end, params_json, run_conf_json, run_type).

XCom chaining between tasks — like Airflow

Tasks pass data the Airflow way: value = ti.xcom_pull(task_ids="upstream"). There is no live TaskInstance in a standalone pod, so the runtime supplies _StandaloneTaskInstance — a tolerant shim whose xcom_pull resolves from the {task_id: value} map the agent delivered in DEXAFLOW_UPSTREAM_XCOM, and whose xcom_push(key, value) captures custom keys for shipping back.

On the control-plane side, the agent only delivers an upstream’s value when the task actually declares the dependency: FetchXCom is guarded by the task’s depends_on list (a real cross-task fetch that the k3d e2e — not the unit tests — first exercised).

For TaskFlow @task callables, _resolve_kwargs injects this same context into named parameters and a **context catch-all, with precedence call-args / XCom > context, so def f(x, ds=None, **context) gets the upstream value for x, the macro for ds, and everything else via context.

Multi-key XCom

Beyond the single return_value, an operator (or a @task via **context) can ti.xcom_push(key="row_count", value=7). The runtime collects these in the ti shim’s pushed dict and writes them to DEXAFLOW_PUSHES_PATH; the agent reads that file and publishes each as an individual XCom. This gives native tasks the same multi-key XCom that operators emit.

Extra-links (the “Open in BigQuery” buttons)

Airflow operators expose operator_extra_links — the deep-link buttons in the UI (open the BigQuery job, the EMR step, the Dataflow graph). The runtime reproduces them generically by calling Airflow’s own link.get_link(operator, ti_key), which reads values the operator stashed via BaseXCom.get_value. The runtime bridges that lookup to the captured pushes (_generic_link_url), with a templated URL fallback (_format_link_url) for older link styles. Computed links are written to DEXAFLOW_EXTRA_LINKS_PATH; the agent ships them as the reserved _extra_links XCom, and the API serves them from the resource’s /links action so the Airflow UI renders the buttons.

Native parity matrix

The new capabilities were standardized so native task types get them too, without changing existing behaviour:

bash gets the context through Jinja rendering of the command ({{ ds }}, {{ var.value.X }}, {{ params.region }}) when the command contains {{ — a plain command stays a direct bash -c so bash-only images need no Python. A templated command that cannot render (a Jinja error, an expression the sandbox refuses, or an image without Jinja2) fails the attempt without running anything, with the error in the task log; the raw template text never reaches the shell (ADR 0084). The base image ships Jinja2 with the Task SDK. Every interpolated value is shell-quoted (shlex.quote) before exec, so a params value from an untrusted conf cannot inject shell — write interpolations unquoted (--name {{ params.x }}, not --name "{{ params.x }}"). See DAG authoring (issue #489).

Security — the auto-quoting is the native bash path only. A captured provider operator (airflow_operator) renders its template_fields with Airflow’s own Jinja, exactly as upstream Airflow does — Dexaflow does not (and cannot, without breaking non-shell operators and Airflow parity) inject shlex.quote into that render. So if you use a provider operator that executes a shell (e.g. a BashOperator) and template an untrusted value into a shell field — params/conf supplied by anyone with execute:dag — the same rules as upstream Airflow apply: quote it yourself, or, preferably, pass the value through the environment ($AIRFLOW_VAR_*, $AIRFLOW_CONN_*, an env kwarg) instead of rendering it into the command string. Trusted, author-written template structure is fine; untrusted interpolated values are the hazard.

Not yet supported (loud, not silent)

Capabilities that need scheduler/control-plane work are rejected loudly rather than half-running:

• Dynamic task mapping — #376.
• Branching — #377.

Reschedule-mode sensors are supported. Set mode="reschedule" and the scheduler persists the next-poke time, frees the pod between pokes, and re-dispatches the sensor preserving its try_number — the idiomatic way to wait in Dexaflow (see DAG authoring).

See also

• ADR 0040 — Airflow operator + sensor execution — the design and the spikes behind it.
• ADR 0024 — parser structural shim — how compile-time capture works.
• Variables & Connections — how secrets reach a task.
• Examples and examples/gcp/ — runnable operator DAGs.