Airflow 3.X connector compatibility

Status: planning document, not yet ADR 0036. The recommendation here is a proposal to feed ADR 0036 before any code lands. Scope is exclusively Airflow 3.X (3.2.x line, main as of 2026-06-02). Airflow 2.x is out of scope.

Goal:

• Long-term: Leoflow stays architecturally independent of Apache Airflow (no Airflow at runtime, no Flask/SQLAlchemy/Pendulum stack in our pods).
• Short-term: existing Airflow 3.X DAGs that use connectors (from airflow.providers.postgres.hooks.postgres import PostgresHook) run on Leoflow unchanged, so adoption costs nothing for current Airflow users.

Both goals are reachable simultaneously with a small Python shim. This document measures how small.

The model in one picture — connection metadata and connector code never live together

Connection metadata (host, login, password, extra JSON) is owned by the Leoflow control plane: created in the admin UI, encrypted at rest (ADR 0019). Connector code (PostgresHook.get_records(), GCSHook.upload(), …) is owned by the user’s DAG image: pip-installed from apache-airflow-providers-<X> declared in leoflow.yaml.dependencies. Leoflow ships no provider code.

The two meet only at runtime, through the AIRFLOW_CONN_<ID> environment variable (ADR 0021 wire format) that the Leoflow agent stamps into the task process. The Leoflow runtime compat shim (ADR 0036) intercepts BaseHook.get_connection() and returns the canonical Connection. The upstream hook does the real work.

┌─────────────────────────────────────────┐    ┌──────────────────────────────────┐
│  Admin UI / API  (Leoflow control plane,│    │  leoflow.yaml  (user, per-DAG)   │
│                   Go — ADR 0014)        │    │                                  │
│                                         │    │  dependencies:                   │
│  POST /api/v2/connections               │    │    - apache-airflow-providers-   │
│  {conn_id: "my_pg", type: "postgres",   │    │      postgres==6.0               │
│   host: "...", login: "...",            │    │    - psycopg2-binary==2.9        │
│   password: "...", extra: "{...}"}      │    │                                  │
└────────────────────┬────────────────────┘    └─────────────────┬────────────────┘
                     │                                            │
                     ▼                                            ▼
┌─────────────────────────────────────────┐    ┌──────────────────────────────────┐
│  Leoflow DB (encrypted, ADR 0019)       │    │  DAG image (built once per push) │
│  connections:                           │    │  ─────────────                   │
│    id="my_pg" type="postgres"           │    │  leoflow-base:py3.11             │
│    password = AES-256-GCM(...)          │    │  + apache-airflow-providers-     │
│    extra    = AES-256-GCM(...)          │    │      postgres   (PostgresHook)   │
└────────────────────┬────────────────────┘    │  + leoflow-runtime-compat-shim   │
                     │                          │      (airflow.sdk.* shim,        │
                     │ on dispatch              │       ADR 0036)                  │
                     ▼                          └─────────────────┬────────────────┘
┌─────────────────────────────────────────┐                      │
│  Leoflow agent (Go, in the pod or       │                      │
│  subprocess host)                       │                      │
│  - decrypts password + extra            │                      │
│  - renders the URI:                     │                      │
│    AIRFLOW_CONN_MY_PG=                  │                      │
│      postgres://login:pw@host:5432/db   │                      │
│      ?__extra__={"sslmode": ...}        │                      │
│    (ADR 0021 wire format)               │                      │
│  - injects env var into the task        │                      │
└────────────────────┬────────────────────┘                      │
                     │                                            │
                     └────────────────► task process ◄────────────┘
                                              │
                                              ▼
                ┌──────────────────────────────────────────────┐
                │  User DAG (Python)                           │
                │                                              │
                │  from airflow.providers.postgres.hooks       │
                │       .postgres import PostgresHook          │
                │  hook = PostgresHook(postgres_conn_id=        │
                │                      "my_pg")                │
                │  hook.get_records("SELECT 1")                │
                └─────────────────────┬────────────────────────┘
                                      │
                                      ▼ provider calls
                                      │ BaseHook.get_connection("my_pg")
                ┌──────────────────────────────────────────────┐
                │  Leoflow runtime compat shim  (ADR 0036)     │
                │  ──────────────────────────                  │
                │  1. Read AIRFLOW_CONN_MY_PG env var          │
                │  2. Parse URI → host/login/password/port/    │
                │      schema/extra                            │
                │  3. Pre-processor (per-type policy seam)     │
                │     • cloud type → cloud resolver            │
                │       (ADR 0035 chain: keyfile_dict →        │
                │        key_path → key_secret_name → ADC)     │
                │       fetches Secret Manager NOW if          │
                │       key_secret_name is set; emits the      │
                │       fetched key as a transient keyfile_    │
                │       dict so the upstream hook understands. │
                │     • non-cloud type → pass through.         │
                │  4. Return a canonical                       │
                │      airflow.sdk.definitions.Connection      │
                └─────────────────────┬────────────────────────┘
                                      │
                                      ▼
                ┌──────────────────────────────────────────────┐
                │  Upstream PostgresHook                       │
                │  (apache-airflow-providers-postgres)         │
                │  - receives the canonical Connection         │
                │  - psycopg2.connect(...) → real query        │
                └──────────────────────────────────────────────┘

What works out of the box vs what needs a provider declared

Two tiers of “Airflow imports that work on Leoflow” — they have very different dependency contracts, and the cookbook pages must state this in the very first line.

Forgetting the dependency for a tier-B import is a fast, loud failure: ModuleNotFoundError: No module named 'airflow.providers.postgres' at import time, before any task work runs. The cookbook page for the hook names the exact error so search engines route the user back to the right recipe.

Out of scope (still rejected at compile time per the closed-set policy): sensors, dynamic task mapping (.expand / .partial), TaskGroup, branching operators, untyped operators outside the standard / http providers. Parser fails fast with “not supported by Leoflow”.

0. TL;DR

Recommendation: Strategy A. ~1,200 LOC of pure Python unlocks ~80 upstream providers without touching ADR 0024’s spirit. The Connection model in Leoflow already has 100% field-level parity with Airflow 3.X’s, so the gap is only “give user code a BaseHook.get_connection()-shaped door into the env vars the agent already injects.”

1. Architecture in Airflow 3.X

A reminder of the 3.X layout, because it differs from 2.x in ways that materially simplify embedding.

1.1 The canonical import paths moved into a task SDK

In Airflow 3.x the runtime hook surface lives at airflow.sdk.*, not at the 2.x paths (airflow.hooks.base, airflow.models.Connection). Provider code no longer imports those 2.x paths directly — they import from a stable compatibility re-export package:

# How a 3.x provider opens — this is the import surface to mimic.
from airflow.providers.common.compat.sdk import (
    BaseHook,
    Connection,
    AirflowException,
    AirflowOptionalProviderFeatureException,
    conf,
)

That airflow.providers.common.compat.sdk is a thin re-export of:

The provider-to-provider contract is the compat layer, not the SDK. That means a shim only has to mock compat.* precisely; the SDK itself we can implement with looser fidelity.

1.2 BaseHook (107 LOC) — what providers actually call

The full public surface a Hook subclass uses from BaseHook:

• __init__(self, logger_name=None)
• get_connection(cls, conn_id) -> Connection (the workhorse)
• aget_connection(cls, conn_id) -> Connection (async siblings call this)
• get_hook(cls, conn_id, hook_params=None) (used by sensors)
• get_conn(self) (abstract — each provider implements)
• get_connection_form_widgets(cls) + get_ui_field_behaviour(cls) (form metadata for the Airflow UI; Leoflow can return {})
• log property (from LoggingMixin)

That’s the entire seven-method surface to fake.

1.3 Connection (571 LOC) — what providers actually use

Despite the file’s size, providers use a narrow subset:

• conn_id, conn_type, host, login, password, port, schema, extra
• extra_dejson property (returns json.loads(extra) or {})
• get_uri() (SQLAlchemy-friendly URI; the inverse of from_uri)
• from_uri(uri) class method (constructs from AIRFLOW_CONN_<ID> value)
• EXTRA_KEY = "__extra__" class constant

The other ~450 LOC handle Fernet roundtripping (not relevant — Leoflow encrypts elsewhere) and the metastore ORM (not relevant — Leoflow’s metastore is Go).

1.4 Connection resolution chain

_get_connection(conn_id) in airflow.sdk.execution_time.context:

1. SecretCache.get_connection_uri(conn_id) → if hit, build Connection.from_uri(...)
2. For each backend in ensure_secrets_backend_loaded():
     conn = backend.get_connection(conn_id) → if non-None, return
3. raise AirflowNotFoundException(...)

The default backend chain in 3.x starts with the EnvironmentVariablesBackend, which reads exactly the AIRFLOW_CONN_<ID> env var Leoflow’s agent already produces (internal/agent/runner.go:170).

Implication: Leoflow’s shim can skip the entire backend chain and read the env var directly. Strictly less code, strictly more deterministic.

1.5 Provider package layout in 3.x

Each provider is a separate PyPI package (Apache 2.0) at providers/<name>/src/airflow/providers/<name>/{hooks,operators,sensors, transfers,triggers}/. Distributed independently (apache-airflow-providers-<name> on PyPI).

The runtime requirement of each provider package on apache-airflow>=3.0 is the compat re-export, not core Airflow itself. Mock compat.* and the hard dep on apache-airflow is satisfied (the package metadata cares about the import surface, not the wheel).

2. Provider catalog (top-15, measured against upstream 3.x main)

For each provider: file path of the main hook(s), LOC, what they import from airflow.*, what underlying SDK they wrap, how thick the wrapper is.

2.1 The hidden lynchpin: apache-airflow-providers-common-sql

Every DB provider (postgres, mysql, sqlite, mssql, snowflake, plus oracle, db2, trino, vertica, exasol, etc.) extends DbApiHook defined in providers/common/sql/src/airflow/providers/common/sql/hooks/sql.py — ~1,250 LOC that depends on airflow.exceptions, airflow.providers.common.compat.module_loading, airflow.providers.common.compat.sdk (BaseHook + conf), and airflow.providers.common.sql.dialects.

This file is the single biggest lever in the whole study: vendor it verbatim (Apache 2.0 permits with attribution) and ~10 DB providers light up at zero re-implementation cost.

2.2 The other lynchpin: airflow.providers.common.compat

compat.sdk, compat.connection, compat.lineage and compat.module_loading are universally imported. They’re each ~10-30 LOC of pure re-exports. Any embedding strategy must mock this package.

3. Three embedding strategies — concrete cost

Strategy A — Stub Airflow core minimally, use upstream providers

We extend the existing parser shim (parser/leoflow_parser/_shim/airflow/) into a runtime shim that provides:

Total shim: ~1,100-1,300 LOC of pure Python that covers ~95% of the surface DB + HTTP + S3 + GCS + Redis + Slack hooks actually touch.

Critical decision: vendor providers.common.sql.hooks.sql.DbApiHook verbatim. Re-distribute the file under Apache 2.0 attribution. That single 1,250-LOC file lights up postgres / mysql / sqlite / mssql / snowflake / oracle / trino / db2 / vertica / exasol at zero re-implementation cost. Same treatment for providers.common.sql.dialects.dialect (~200 LOC).

Image size impact: ~1,500 LOC of pure Python + the user-installed provider wheels. No apache-airflow install. A typical postgres-only image goes from ~80 MB (Python slim) to ~140 MB (slim + psycopg2-binary + sqlalchemy + shim). Versus ~600 MB if Strategy C lets apache-airflow get pulled in.

Risk surface: every Airflow 3.x minor release can rename or add an internal helper. Empirical pattern from 2.x→3.x: ~3-5 breaking internal moves per minor. Maintenance: budget 1-2 engineer-days per Airflow minor to keep the shim aligned, gated by a CI matrix that pip-installs each upstream provider against the shim and runs its smoke tests.

ADR 0024 alignment: the existing shim is parser-only. Extending it to a runtime shim is a deliberate, principled scope expansion of the same idea (“we never install Airflow”), not a contradiction. Worth a new ADR (“Runtime hook compatibility shim”).

Strategy B — Re-implement hooks natively in leoflow_runtime.hooks.*

Estimated LOC for native re-implementations of the top-15 (Leoflow-flavored, no Airflow surface):

Plus a shared leoflow_runtime.connections.Connection (~150 LOC) and BaseHook (~80 LOC): ~4,000 LOC total.

Pros: owns the surface; ADR-clean; no Airflow version drift; full control over error messages, logging, and observability.

Cons: the upgrade story is brutal. Every Airflow DAG that does from airflow.providers.postgres.hooks.postgres import PostgresHook breaks. We’d need an import-rewriter (airflow.providers.postgres.hooks.postgres → leoflow_runtime.hooks.postgres) and the public surface would need to match Airflow’s method signatures anyway — most of the LOC savings vanish.

Strategy C — Allow user-installed apache-airflow-providers-*

User declares apache-airflow-providers-postgres in leoflow.yaml.deps; runtime pip-installs at venv build / image build time. The providers import from airflow.providers.common.compat.sdk import BaseHook, Connection which forces apache-airflow>=3.0 as a runtime dep.

• Image size impact: apache-airflow 3.2.x is ~150-180 MB installed (Flask, SQLAlchemy, Alembic, Pendulum, FAB, Connexion, …). On top of slim Python: image grows from ~80 MB → ~280-350 MB before the provider’s own deps. With postgres + psycopg2-binary + sqlalchemy, expect ~450 MB. A real-world DAG image with 3 providers: 600-800 MB.
• Conflict with ADR 0024: the ADR forbids importing real Airflow during parsing. It does not formally forbid importing it at runtime. However: (a) it makes the Lite edition’s subprocess executor terrible (cold start dominated by Airflow’s import of Flask + SQLAlchemy + Pendulum, easily 4-7 s on a laptop); (b) it pulls in CVEs we can’t audit; (c) it imports a database stack into a task pod that has no DB.
• Conflict with “Python minimal, Go max” principle: very large negative impact. apache-airflow brings ~600 transitive Python deps.

Verdict: acceptable only as an escape hatch for connectors we don’t support; never the default path.

4. Connection model gap (field-by-field)

Already 100% parity. The Leoflow Connection model (internal/domain/connection.go) and the AIRFLOW_CONN URI renderer (internal/storage/conn_uri.go:40) cover every field a 3.x Connection exposes:

URI form <conn_type>://<login>:<password>@<host>:<port>/<schema>?__extra__=<json> — Leoflow already renders this byte-for-byte. The only edge case handled: sqlite:/// triple-slash idempotency (conn_uri.go:31-35).

Gap on the model side: zero. The Connection is feature-complete for Airflow 3.X parity. The gap is entirely in how that connection is consumed: today user code reads the env var and parses the URI manually. Once a BaseHook.get_connection() exists, every Airflow-style hook works.

5. Recommendation

Pick Strategy A as the default. Use Strategy B as a targeted overlay for the 5 hooks where Leoflow-native ergonomics matter.

Why:

1. The shim already exists in spirit (the parser shim at parser/leoflow_parser/_shim/airflow/) and ADR 0024 already documented the “shim, don’t install” principle. This is an extension, not a reversal.
2. Field-level parity on Connection is already 100%. The only missing piece is a BaseHook class wired to the existing AIRFLOW_CONN_* env-var delivery — which Leoflow’s agent (internal/agent/runner.go:170) already produces.
3. ~1,200-1,500 LOC of Python shim unlocks the long tail (~80 providers). Re-implementing all 80 natively would be 15k+ LOC and nobody on Leoflow’s roadmap wants to maintain that.
4. The 3.x design — secrets backends, no Fernet in the SDK, attrs-based dataclasses — is dramatically more shimmable than 2.x’s airflow.models jungle. The window to do this cheaply is now; only 3.x as the cut matters.

Phasing

Phase A.0 — Foundations (~1 sprint, ~1-2 weeks of focused work).

• Write a failing integration test:

  from airflow.providers.postgres.hooks.postgres import PostgresHook
  rows = PostgresHook(postgres_conn_id="my_db").get_records("SELECT 1")
  

  inside a real Lite-executed task. This is the regression contract.
• Move the parser shim into a shared package: leoflow_python_compat/airflow/... with two entry points (parser-mode = lazy stubs; runtime-mode = real shim). Runtime install gated by leoflow.yaml.airflow_compat: true.
• Implement BaseHook, Connection, Variable, exceptions, log.mask_secret, execution_time.context._get_connection (reads AIRFLOW_CONN_* env directly — skip the SecretsBackend chain entirely; it’s overkill for our model).
• Vendor providers.common.sql.hooks.sql.DbApiHook + providers.common.sql.dialects.dialect verbatim (Apache 2.0 attribution).
• CI matrix: pip-install upstream apache-airflow-providers-{postgres,sqlite,redis,http} against the shim and run their unit tests. This is the regression gate.

Phase A.1 — The 80/20 cut. Ship official Leoflow support (CI matrix + cookbook page + connection-test DAG, matching the precedent set by the prior connector-rigor work) for these five hooks:

1. postgres — Lite’s own datastore tier; the most-used connector full-stop.
2. http — REST APIs, webhooks; the lingua franca of integration.
3. sqlite — 51 LOC upstream, almost free; great for tutorials.
4. redis — 162 LOC, ~zero risk; already a Leoflow infra primitive.
5. mysql — Postgres’ counterpart; very common.

These five cover the majority of “I want to migrate my Airflow DAG” cases per Airflow’s own provider download stats (postgres + http + mysql consistently top-3 downloads).

Phase A.2 — Cloud expansion. Add S3, GCS, WASB. These three account for almost all object-storage traffic. Each is heavy upstream (1k+ LOC) but our cost is zero if Strategy A holds — we just need the shim to satisfy their imports.

Phase A.3 — Long tail. Slack, FTP, SSH, Snowflake, Databricks, dbt: documented as “should work via the shim, not formally tested.” Users opt-in via airflow_compat: true.

Phase B (later, optional overlay). For the 5 hooks in A.1 only, ship a native leoflow_runtime.hooks.postgres.PostgresHook that subclasses the upstream PostgresHook by composition. This gives Leoflow-native ergonomics (better error messages, structured logging into our lifecycle stream) while still satisfying isinstance(h, PostgresHook) for DAG code. Total new LOC for B: ~500. Skip this until A is in production for 2 releases.

Risks

1. Airflow provider drift. A minor release renames providers.common.compat.sdk. Mitigation: CI matrix on every Airflow minor; pin the supported Airflow line per Leoflow release (e.g. Leoflow 0.2 supports Airflow providers compatible with apache-airflow~=3.2, 0.3 bumps to 3.3). Document the support matrix.
2. License. All providers are Apache 2.0 — re-distributing DbApiHook is explicitly permitted with attribution. No GPL anywhere in the top-15.
3. Provider-to-provider deps. PostgresHook imports AwsBaseHook and AzureBaseHook for IAM. The cleanest workaround: monkey-patch those imports to None-yielding stubs and lazily fail only when the user actually invokes IAM auth. Cost: ~30 LOC of import hooks. Same trick handles OpenLineage cross-imports.
4. Async surface. Async hooks (aget_connection, HttpAsyncHook) use sync_to_async from asgiref and asyncio. The shim must stub _async_get_connection. Trivial — already shown in 3.x source.
5. Connection encryption mismatch. Leoflow encrypts password at rest; Airflow 3.x SDK does not. Zero conflict — encryption is metastore- side. We decrypt before stamping AIRFLOW_CONN_* (already the case at internal/agent/runner.go:170).
6. Variable scope. Leoflow Variables are plaintext today. The shim’s Variable.get() will read AIRFLOW_VAR_<KEY> which the agent already emits (internal/agent/runner.go:163) — no behavior change required.

Bottom line

Build a ~1,200 LOC Python “runtime compatibility shim” that fakes airflow.sdk.definitions.{hooks.base, connection, variable}, airflow.sdk.exceptions/log/execution_time.context, the airflow.providers.common.compat.* re-export layer, and vendors the 1,250-line DbApiHook verbatim. Gate it behind an opt-in airflow_compat: true in leoflow.yaml so the default Lite footprint stays slim. Officially support 5 hooks via CI (postgres / http / sqlite / redis / mysql), document the rest as best-effort. The Connection model has zero gap. Avoid Strategy C (pulling apache-airflow into the image) at all costs — it adds 200+ MB and 600 transitive Python deps for no architectural gain. Strategy B (native re-implementation) is a tempting overlay later but the wrong primary path: it would force re-litigating every Airflow method signature for ~4,000 LOC of ongoing maintenance, with no compatibility upside the shim doesn’t already provide.

6. Open questions for the ADR

These are the decisions that ADR 0036 would need to lock before code lands. Not blockers for this study, but listed so they’re explicit:

1. Module namespace. Do we expose the shim at airflow.* (so user imports work unchanged) or at leoflow_compat.airflow.* (so it’s explicit)? Strong lean: airflow.* for compatibility, but only on sys.path when airflow_compat: true. This avoids polluting non- compat builds.
2. Versioned support matrix. Leoflow 0.2 supports providers from Airflow 3.2; 0.3 from 3.3. Or do we pick a single “supported Airflow range” per Leoflow release and document it?
3. Test isolation. Should the CI matrix run upstream provider test suites (slow, large, real network) or only a curated subset of “smoke” tests we write ourselves?
4. Error surface. When an unsupported provider’s hook raises an internal-looking error (e.g. AirflowOptionalProviderFeatureException), do we re-raise as a Leoflow-branded error or let it bubble?

7. Files this study referenced (Leoflow side)

• Parser shim: parser/leoflow_parser/_shim/airflow/_core.py, parser/leoflow_parser/_shim/airflow/sdk/__init__.py
• Runtime: runtime/python/leoflow_runtime/runner.py, runtime/python/leoflow_runtime/xcom.py
• Connection model: internal/domain/connection.go
• AIRFLOW_CONN URI rendering: internal/storage/conn_uri.go
• Agent env injection: internal/agent/runner.go (lines 153-170)
• Encryption ADR: docs/adr/0019-secret-encryption-at-rest.md
• Parser shim ADR: docs/adr/0024-dag-parsing-structural-shim.md