Airflow integration

Airflow is an orchestration adapter around framework dispatch. Step code never imports Airflow.

Manifest-driven DAG generation

drtoller_step_runner.py scans configured roots, loads only step manifests, deduplicates by step_id, and creates one unscheduled DAG for each manifest with orchestrator.dag_id.

Each DAG contains one PythonOperator task (run_step), uses max_active_runs=1, and binds a fixed step id. dag_run.conf.step_id may be omitted; if present and different, execution fails.

orchestrator:
  dag_id: normalize_v1
  step_id: normalize
  connections: [minio_s3, postgres_app]
  artifacts_root: env:DRTOLLER_ARTIFACTS_ROOT
  retries: 0
  retry_delay_s: 60

retries / retry_delay_s apply to the Airflow PythonOperator (defaults 0 / 60). Leave retries at 0 unless the task is safe to repeat — an OOM inflate must not run three times. Set retries: 3 when you want MinIO/DNS blips after a compose recreate to recover.

Step discovery

Priority:

1. DRTOLLER_STEPS_ROOTS_FILE — newline-separated roots;
2. DRTOLLER_STEPS_ROOTS — colon-separated roots;
3. DRTOLLER_STEPS_ROOT — one root;
4. legacy development mounts/fallbacks.

A direct <root>/<step_id>/<step_id>.drtml lookup is attempted first, then recursive discovery validates manifest.step_id.

Triggering a run

build_trigger_conf() sends a minimal payload: step_id, normalized run_id, artifacts_root, and non-empty user fields. client.trigger_dag_run() unpauses the DAG (PATCH is_paused: false) then posts to Airflow REST /api/v1/dags/<dag_id>/dagRuns.

The same trigger is used by drtoller run --via airflow, POST /v1/runs with via: airflow, and Streamlit. Worker-side merge is prepare_run_from_conf (same hooks as local CLI).

Environment:

• AIRFLOW_BASE_URL
• AIRFLOW_USERNAME (or AIRFLOW_USER)
• AIRFLOW_PASSWORD

Credentials should be provided by deployment secrets; defaults are development-only.

Worker-side parameter construction

run_task builds the worker RunContext through prepare_run_from_conf (same merge as local CLI / HTTP via=local):

1. discovers and loads the manifest;
2. removes reserved step_id, run_id, artifacts_root keys;
3. merges user overlay with params_defaults;
4. calls optional orchestrator.params_hook;
5. applies orchestrator.env_fallbacks/env_fallback_unless;
6. mirrors run id into runtime.run_id and, for legacy mapping mode, mapping.run_id when absent;
7. applies declared connection hooks.

The resulting RunContext is passed to processing.dispatch.run_from_manifest(). If the manifest lists minio_s3, the task waits until MINIO_ENDPOINT hostname resolves (wait_endpoint_resolvable) before dispatch — Docker DNS after Airflow recreate.

Connections

orchestrator.connections declares external connection ids. Current builtin MinIO handling reads Airflow connection minio_s3 and translates login/password/extras into the storage S3 environment.

PostgreSQL and Qdrant transports resolve their named DRTML connections through framework DB integration. New Airflow connection types should be generic adapters keyed by declared connection id, never branches on step id.

Params hooks

orchestrator:
  params_hook: acme_pipeline.normalize.params:complete_params

Signature:

def complete_params(manifest, params: dict, user_params: Mapping) -> dict:
    ...

Use it only to complete integration/runtime configuration. Domain policy should stay in the processor and explicit domain.* defaults.

XCom

Framework always pushes <step_id>_outputs from result.outputs (or the result itself). Extra values use declarative dot paths:

orchestrator:
  xcom:
    - key: manifests
      from: outputs.storage_manifests
    - key: documents
      from: outputs.docs_total

Missing path segments resolve to null rather than executing code.

Failure behavior

Manifest discovery/validation, fixed-step mismatch, connection lookup, params hook, dispatch, or runtime exceptions fail the Airflow task. Framework logs a bounded result preview after success.

Extending Airflow

Add external-system translation under integration/airflow; keep DAG objects and hooks out of storage/processing/domain layers. Test generated DAG identity, minimal conf, merge order, connection translation, and XCom resolution independently.