Formal Airflow provider and lightweight pack reader¶

Purpose: install and operate the parse-safe dpone provider in Airflow without pulling connector/runtime dependencies into the control plane.

Audience: Airflow platform engineers, DAG repository maintainers, and operators diagnosing provider parse failures.

Install apache-airflow-providers-dpone in Airflow scheduler, DAG processor, API server and worker environments. It owns provider discovery and the typed airflow.providers.dpone namespace. Its dependency dpone-airflow-pack is the Airflow-independent static pack reader and DAG construction library.

It is intentionally separate from the full dpone runtime:

Layer	Package	Responsibility
Airflow scheduler / DAG processor / API server	`apache-airflow-providers-dpone` + `dpone-airflow-pack`	Discover the provider, parse static packs, read bounded cache and build visible KPO/outcome tasks.
KPO runtime pod	`dpone[full,accel]`	Execute manifests, hooks, SQL transforms, transfer, lineage, DQ, audit and cleanup.

Install a supported scheduler image¶

Use a tested cell from the Airflow compatibility matrix. This executable example selects the primary Python 3.12 / Airflow 3.2.0 / Kubernetes provider 10.14.0 cell. Following Apache Airflow's installation guidance, use the official constraints file only for the Airflow installation, then pin Airflow again while adding the dpone provider:

AIRFLOW_VERSION=3.2.0
PYTHON_VERSION=3.12
AIRFLOW_CONSTRAINTS_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"

python -m pip install \
  "apache-airflow[cncf.kubernetes]==${AIRFLOW_VERSION}" \
  --constraint "${AIRFLOW_CONSTRAINTS_URL}"
python -m pip install \
  "apache-airflow==${AIRFLOW_VERSION}" \
  "apache-airflow-providers-cncf-kubernetes==10.14.0" \
  "apache-airflow-providers-dpone==0.73.20"
python -m pip check
python -c "from airflow.providers.dpone import load_dpone_dags; print(load_dpone_dags.__module__)"

The provider distribution declares the compatible dpone-airflow-pack dependency. Resolve the image once, record the exact installed versions and wheel digests, then use that immutable image in every Airflow component that imports DAG files. Keep the authoring environment and KPO runtime image separate: authoring needs base dpone, while connector extras and native clients belong in the runtime pod.

Recommended DAG import¶

from airflow.providers.dpone import load_dpone_dags

load_report = load_dpone_dags(
    globals(),
    index_path="/opt/airflow/dags/.dpone-cache/current/airflow-index.json",
)
if load_report.fatal:
    error_code = (
        load_report.errors[0].get("code", "DPONE_AIRFLOW_INDEX_INVALID")
        if load_report.errors
        else "DPONE_AIRFLOW_INDEX_INVALID"
    )
    raise RuntimeError(
        f"{error_code}: dpone Airflow deployment index could not be loaded"
    )

For legacy provider imports, generated loader upgrades, or the mutable latest/pack-index.json cache, follow the provider and cache migration guide.

Indexed delivery boundary after v0.73.1¶

The published v0.73.1 line is the compatibility baseline. The approved hardening patch after v0.73.1 introduces a separate executable v2 index; do not hand-edit a v0.73.1 v1 index and assume it has acquired v2 guarantees. Confirm that the producer, scheduler provider, and runtime image all support the same v2 contract before promoting it.

The provider evaluates the wire and delivery mode before adding any DAG to the module:

Index wire	Mode	Provider behavior
`dpone.airflow-deployment-index.v1`	`local_preview`	Preserve the non-runnable local preview lane.
v1	`init_fetch`	Fatal `DPONE_RUNTIME_ARTIFACT_DELIVERY_MIGRATION_REQUIRED`; regenerate and promote a v2 deployment.
`dpone.airflow-deployment-index.v2`	`init_fetch`	Build the strict executable KPO path.
v2	Any other known mode	Fatal `DPONE_RUNTIME_ARTIFACT_DELIVERY_MODE_UNSUPPORTED`.
Either	Unknown or malformed mode	Fatal index field/schema error.

The executable path is:

producer -> immutable release/deployment -> airflow-index.json (schema v2)
-> trusted local cache -> lightweight provider -> KPO
-> runtime-init-fetch -> runtime-fetch-ready.json
-> runtime-pack-exec -> verified workload argv

V2 requires an exact OCI image reference whose digest matches runtime_image_digest, positive byte counts and SHA-256 for the exact release/deployment/workload artifacts, workload identity, and digest-pinned registry/trust-policy ConfigMap references. trust_tier is explicit at both index and delivery levels and must match. production requires trust_policy_ref plus attestations: required_for_prod; without a concrete runtime verifier, init fails before registry I/O.

Every strict-v2 workload pack also declares pack_identity.schema: dpone.airflow-pack-identity.v1. Its pack_fingerprint is derived from the complete canonical JSON pack, excluding only the self-reference and the fixed top-level advisory metadata allowlist. The producer, deployment materializer, provider and runtime independently derive that fingerprint. Comparing the index claim with a copied pack claim is not verification. A legacy pack without this identity remains readable only through the explicit local/raw compatibility lane and cannot become an executable strict-v2 deployment.

The stock runtime CLI in this patch intentionally does not ship a concrete production attestation verifier. Therefore strict-v2 production execution remains fail-closed and UNVERIFIED; do not promote it as a runnable production path until a separately approved verifier contract, implementation, and live Kubernetes evidence are available. Non-production checksum/receipt verification does not substitute for provenance attestation.

The provider creates a canonical plan of at most 16 KiB, carries it in DPONE_INIT_FETCH_PLAN_B64, and pins its digest in DPONE_INIT_FETCH_PLAN_SHA256 and the pod annotation. It replaces pack-owned execution fields with:

init container: dpone airflow runtime-init-fetch
base container: dpone airflow runtime-pack-exec

Both containers use the exact digest-pinned image. The fixed storage contract is:

Volume	Init	Base
`dpone-fetched-artifacts`	RW `/var/lib/dpone/artifacts`	RO `/var/lib/dpone/artifacts`
`dpone-worktree`	RW `/workspace/repo`	RO `/workspace/repo`
`dpone-run-output`	absent	RW `/var/lib/dpone/run`
`dpone-artifact-registry-config`	RO `/etc/dpone/artifact-registry`	absent
`dpone-artifact-trust-policy`	RO `/etc/dpone/artifact-trust` when selected	absent

Image, commands, arguments, namespace, service account, security context, reserved volumes/mounts, and init containers are provider-owned. Pack scheduling and resource fields remain the bounded extension surface. The pack must carry the exact dpone.airflow-provider-execution.v1 projection and a digest-pinned XCom sidecar image; missing, unknown or execution-owning fields fail before DAG installation. See the v2 wire ADR, the runtime operator runbook, and the migration procedure.

The current TCB includes the producer, promoted cache/index, scheduler/DAG processor, provider, Kubernetes control plane and kubelet, runtime image, workload-identity enforcement, registry adapter, and configured verifier. The fixed composition prevents an ordinary pack override; it does not claim resistance to a compromised scheduler, provider, or local cache. Live Kubernetes, workload identity, attestation, Vault, MSSQL, and ClickHouse remain UNVERIFIED without current evidence from the exact commit and environment.

DponeDag.from_spec(...) and DponeTaskGroup.from_pack(...) remain the hybrid escape hatches. Direct dpone_airflow_pack loaders are legacy pack-generation compatibility paths, not the release/deployment delivery API:

from dpone_airflow_pack import load_dpone_airflow_pack_with_provenance

pack, provenance = load_dpone_airflow_pack_with_provenance("cached://inter_ch_kontr_inn_for_email")

provenance includes the cache source, generation, pack checksum, index checksum and last sync status. This is the right evidence to serialize into Airflow task metadata. KubernetesExecutor worker pods do not need to mount the same cache volume as the DAG processor.

A direct file-path/raw-pack DponeTaskGroup.from_pack(...) call has no v2 delivery context and cannot claim strict init_fetch. A deployment-scoped cached:// workload resolved through a v2 index_path receives that index's immutable delivery context and uses the strict path.

Environment deployment promotion uses the separate content-addressed release/deployment cache contract. Platform operators should follow the Airflow cache sync and recovery runbook for pinned release verification, current-pointer promotion, structured checksum errors, and actor/CAS-bound recovery after a partial local write. Promotion uses a local inter-process lock and activates the relative current symlink only after pointer/audit authorization. Provider parsing never performs that sync, locking, or recovery work. The deployment-index reader requires canonical lowercase sha256:<64 hex> release, deployment, binding, registry, credential runtime, image, and cached-reference pin identities, matching the public JSON Schema exactly.

GitOps Domain Groups¶

DAG wrappers should not duplicate workload lists when the same order is already declared in GitOps catalogs. Keep repeatable orchestration groups in the domain YAML:

workflow_groups:
  daily_datamarts:
    workload_ids:
      - dim_customer
      - fact_order
      - fact_order_history

Then resolve the ordered list with the lightweight provider:

from dpone_airflow_pack import workload_ids_from_gitops_domain

workload_ids = workload_ids_from_gitops_domain(
    repo_root,
    "sales",
    group="daily_datamarts",
)

The resolver is scheduler-safe: it reads only local YAML files, checks that referenced workloads exist in the same domain, and has no dependency on full dpone, database drivers, Kubernetes clients or cloud SDKs. This keeps the Airflow wrapper thin while making GitOps catalogs the single source of truth for workload membership.

dpone.airflow.* remains a compatibility shim when full dpone is installed. Scheduler-internal library helpers may use dpone_airflow_pack directly so the lightweight package stays independent from source/sink runtime dependencies. DAG files and provider-facing code use only airflow.providers.dpone.

Provider-facade imports from dpone_airflow_pack are deprecated compatibility exports. They remain available for at least two minor releases and 365 days after the 0.72.0 announcement, whichever is later, and warn at most once per process. New DAG code always imports airflow.providers.dpone.

Declarative loader (`load_dpone_dags`)¶

For YAML-only DAG repos, prefer the canonical loader:

from airflow.providers.dpone import load_dpone_dags

load_report = load_dpone_dags(
    globals(),
    index_path="/opt/airflow/dags/.dpone-cache/current/airflow-index.json",
)
if load_report.fatal:
    error_code = (
        load_report.errors[0].get("code", "DPONE_AIRFLOW_INDEX_INVALID")
        if load_report.errors
        else "DPONE_AIRFLOW_INDEX_INVALID"
    )
    raise RuntimeError(
        f"{error_code}: dpone Airflow deployment index could not be loaded"
    )

The loader materializes DAG objects from fingerprinted dag-spec artifacts and compact packs in the bounded cache. It skips dag_ids already present in globals() for collision-safe migration. Invalid specs use skip_and_report by default; paused placeholder DAGs tagged dpone_spec_error require the explicit create_diagnostic_dag platform policy.

The index itself is a shared trust boundary. The provider reads at most 8 MiB before JSON parsing; a missing, unreadable, oversized, malformed, or contract-invalid index returns LoadReport(fatal=True) under the default policy. The generated loader guard above raises so Airflow cannot present an empty parse as success. invalid_dag_policy="fail_all" raises the original AirflowDeploymentIndexError directly.

Listed artifacts are isolated after the index is accepted. Each DAG spec is read once and checked against its declared byte size and SHA-256 before parsing; each workload pack is checked again at its final task-building boundary. One missing, oversized, size-mismatched, or checksum-mismatched artifact is reported with its dag_id while unrelated valid DAGs continue under skip_and_report. The default artifact read ceiling is 64 MiB. Restore or rematerialize immutable content from the trusted release; never edit a listed cache file in place. See Provider API: fatal index errors and recovery.

Producer outlets are declared under airflow.execution.outlets in workload catalogs. Consumer DAGs may use dag-spec schedule.assets or manifest airflow.execution.inlets. Build plane resolves cross-workload lineage; review with:

: "${DPONE_WORKLOAD_SET:?set the repository-relative workload-set path}"
dpone gitops airflow deps --workload-set "${DPONE_WORKLOAD_SET}" --format md

See Airflow self-service.

Bounded backfill mapping¶

Generated packs may contain a dpone.airflow-mapping-plan.v1 for one chunked backfill process. internal preserves the existing concrete runtime task. visible and summary use a static list with KubernetesPodOperator.partial(...).expand(env_vars=...); no upstream discovery task or scheduler-side planner is created.

The provider validates the complete mapping plan before constructing an operator: canonical fingerprint, ordered coverage, item count, pool, max_active_tis_per_dag, and the 200-item hard ceiling. Provider-owned pool and concurrency values cannot be replaced by pack overrides. Each mapped task gets one bounded DPONE_AIRFLOW_MAPPING_ITEM value plus the same pinned composite run identity. The item contains only indexes and digests, never predicates, credentials or row data.

Mapped pack bytes use mapped_kpo_kwargs as a compatibility tripwire. This provider validates the plan and normalizes the field internally; providers released before bounded mapping fail their existing missing-kpo_kwargs check before creating any task. Do not hand-edit an immutable pack to rename the field.

Mapped tasks validate their own outcome inline. A shared downstream outcome gate is intentionally omitted because it cannot represent every map index consistently across the supported Airflow lines. Hooks remain static. A provider without compatible partial/expand support fails with DPONE_AIRFLOW_MAPPING_UNSUPPORTED; it never silently falls back to an unbounded or whole-campaign task.

This feature preserves the parse-side-effect contract: only the verified local deployment index and listed artifacts are read. Exact construction tests run against Airflow 2.10.5, 2.11.0, 3.2.0 and 3.3.0 in the provider compatibility workflow. Runtime multi-pod certification remains a separate route/state evidence requirement.

Composite run identity¶

For every index-backed workload task, the provider derives one bounded dpone.airflow-run-identity.v1 object from the already verified local index, DAG spec, and workload pack. It attaches the DAG-level context for diagnostics, stores the workload-specific value in task params, and passes canonical JSON in DPONE_AIRFLOW_RUN_IDENTITY to the runtime pod.

The identity pins release, deployment, DAG spec, workload pack, runtime image, binding-set, connection registry, credential runtime, and Airflow DAG Bundle. It contains no secret values, Vault paths, signed URLs, or authorization material. Operator overrides cannot replace the provider-owned value, and the provider verifies the pack checksum again after reading it to close a local index/file replacement race.

Runtime XCom and gitops.airflow_evidence_bundle preserve the same object. Conflicting observations or an observed pod image mismatch fail closed with DPONE_AIRFLOW_RUN_IDENTITY_MISMATCH. Release-policy evidence without the identity fails with DPONE_AIRFLOW_RUN_IDENTITY_MISSING; older advisory evidence remains readable with an explicit warning.

The identity does not change the parse-side-effect contract. Construction reads only the bounded local deployment index and listed artifacts. It performs no network, metadata DB, Variable, Connection, Vault, Kubernetes, or cache-refresh operation. Use dpone airflow rerun-plan outside DAG parsing to select original or latest Airflow delivery independently from original or latest dpone artifacts.

Credential resolution remains a runtime concern after the provider has pinned the deployment. latest + workload_start resolves once for each workload; unsupported pinned/DAG-wide policies and ambiguous KV v2 version evidence fail closed. Operator recovery and resolver certification status are documented in Airflow credential resolver lifecycle.

After task execution, the evidence collector joins the immutable identity with the concrete Airflow attempt, dpone run/evidence, and observed pod into dpone.airflow-correlation.v1. The attempt digest is stable across late pod observations but changes for a retry or mapped index. Release-policy evidence requires a complete correlation; advisory/PR evidence may carry an explicit DPONE_AIRFLOW_CORRELATION_INCOMPLETE warning. A digest, DAG, pack, or runtime image contradiction always blocks evidence.

Use the final evidence bundle as the optional local input to dpone ops lineage-export --airflow-evidence-bundle ... and dpone observability metrics-export --airflow-evidence-bundle .... Neither exporter is imported by the provider or performs parse-time I/O. OpenLineage gets a versioned custom run facet and deterministic UUIDv5 run ID; OTel gets resource/data-point attributes; Prometheus labels remain unchanged.

Interval-aware runs and data-aware scheduling¶

Compact packs are interval-aware out of the box:

kpo_kwargs.env_vars carries Jinja-templated DPONE_DAG_ID, DPONE_DAG_RUN_ID, DPONE_TRY_NUMBER, DPONE_LOGICAL_DATE, DPONE_INTERVAL_START, DPONE_INTERVAL_END, and the optional DPONE_PARTITION_KEY. Airflow renders them per task instance, so every pod receives the concrete DAG-run interval while the pack stays a static artifact.
Inside the pod, dpone run consumes the same variables: they feed run-state identity (dag_id + execution_date) and resolve {{ data_interval_start }} / {{ data_interval_end }} / {{ ds }} tokens inside manifests. With mode: backfill + inner_mode: partition_replace this makes catchup=True, task clears and airflow dags backfill idempotent per interval — see the Backfill guide and examples/dags/.
Declare produced datasets under airflow.execution.outlets (list of asset URIs) in the workload catalog; the provider converts them into Asset (Airflow 3) / Dataset (Airflow 2.4+) outlets on the runtime task for data-aware downstream scheduling. Images without the Asset API skip outlets safely.
The XCom summary includes the run interval and a bounded backfill progress section (see docs/schemas/gitops/airflow-xcom-summary.schema.json).

Partitioned dag-specs use public Airflow 3.2+ SDK timetables:

a cron producer uses CronPartitionTimetable;
an asset consumer uses PartitionedAssetTimetable with IdentityMapper;
the attached DAG partition plan supplies runtime context even when the consumer inherited the partition and its compact pack has no duplicate declaration;
Airflow 3.0/3.1 and 2.x retain the existing schedule and set DPONE_PARTITION_MODE=degraded_unpartitioned.

The provider fails closed when DAG and pack partition identities disagree or a native task has no scheduler partition key. Capability negotiation imports only airflow.sdk; it performs no parse-time network, metadata, Variable, Connection, Vault, Kubernetes, or cache-refresh access.

Compatibility with Airflow 2.10 LTS and 3.x is exercised in CI (.github/workflows/airflow-pack-compat.yml).

Declarative DAG timezone (Airflow 2+)¶

Domain catalogs and *.dag-spec.json may declare timezone next to a cron schedule. The loader never forwards that field as a raw DAG() kwarg (Airflow 3 rejects it). Instead dpone_airflow_pack.dag_schedule applies the timezone in a version-aware way:

Airflow line	Cron + timezone behavior
2.2+ and 3.x	`CronTriggerTimetable("0 7 * * *", timezone="Europe/Moscow")`
2.0–2.1 (legacy)	`DAG(timezone=..., schedule_interval="0 7 * * *")` when timetables are unavailable
all supported	`start_date` is materialized in the catalog timezone via pendulum

Manual/asset schedules ignore timezone on the schedule itself; only start_date is localized. Invalid timezones quarantine the dag-spec with dpone_spec_error instead of breaking the whole loader module.

Scheduler guardrails¶

The provider performs only deterministic parse-time work:

reads one local deployment index with an 8 MiB default ceiling;
reads only listed DAG specs and workload packs, each with a 64 MiB default ceiling and declared size/SHA-256 verification;
completes the v2 delivery and pod-contract preflight for every listed DAG before publishing any of them to globals();
validates pack kind, KPO kwargs, pod spec and outcome-gate metadata;
builds visible Airflow tasks;
reads Airflow connections only inside task execution when an unsafe development bridge is explicitly enabled.

It does not parse manifests, call databases, call Kubernetes APIs, generate GitOps artifacts, import ClickHouse/MSSQL connectors, or execute business logic.

Packaging¶

Install apache-airflow-providers-dpone==0.73.20 in the base Airflow image by using the constrained, matrix-pinned sequence above. It pulls the matching lightweight reader without the full dpone runtime. Provider and generated runtime packs must stay on the same dpone release line.

Do not install dpone, dpone[full], dpone-native-accel, pyodbc, pandas, polars, ConnectorX or sink/source connectors into the scheduler image unless your platform intentionally runs pipeline runtime code in the scheduler, which is not the recommended production topology.

Product pattern¶

This follows the same separation used by mature orchestration stacks: Airflow provider packages keep scheduler parsing small, Cosmos consumes compiled dbt artifacts instead of parsing projects in the scheduler, and connector-heavy runtimes execute in isolated worker images.

Legacy pack cache CLI¶

This CLI consumes a mutable latest/pack-index.json for pre-release/deployment installations. It remains supported for compatibility, but new installations must use dpone airflow publish, dpone airflow cache-materialize, dpone airflow cache-sync, and the local airflow-index.json provider loader. The legacy sync is pack-only: it is not allowed to manufacture a release-set, deployment-set, or current pointer.

For scheduler startup or a refresh sidecar:

dpone-airflow-pack-sync \
  --once \
  --index-uri s3://bucket/dpone-artifacts/prod/repo/latest/pack-index.json \
  --reader-connection-id s3_dpone_artifacts_reader \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --status-path /opt/airflow/dags/.dpone-cache/airflow/status/last-sync-status.json

For diagnostics:

dpone-airflow-pack-cache-status \
  --cache-dir /opt/airflow/dags/.dpone-cache/airflow \
  --json

The sync command writes success and warning status files. A platform can wrap it fail-open in init containers or sidecars so object-storage failures do not stop ordinary Airflow DAG parsing.

The canonical release/deployment materializer is fail-closed and exact-ID. It runs outside DAG parsing and does not share this command's latest semantics. Use the provider and cache migration guide to cut over without making the scheduler fetch remote state, then follow Airflow cache sync and recovery for ongoing operation.