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

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() posts it to Airflow REST /api/v1/dags/<dag_id>/dagRuns.

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.build_params_from_conf():

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().

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.output_manifests_json
    - key: rows_written
      from: stats.rows_written

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.