Switch the default async Postgres driver from asyncpg to psycopg3 (core)

[Dev-iL]

depends on:

• Make PostgreSQL SQLAlchemy driver explicit (postgresql+psycopg2://) #68314
• Switch the default async Postgres driver from asyncpg to psycopg3 (provider) #69089

closes: #67801

What

When [database] sql_alchemy_conn_async is not set, Airflow derives the async metadata-database URL from sql_alchemy_conn. For PostgreSQL, the derived URL now uses psycopg3 (postgresql+psycopg_async://) instead of asyncpg (postgresql+asyncpg://).

Packaging follows: apache-airflow-providers-postgres now installs psycopg[binary] by default, and asyncpg moves to a new opt-in asyncpg extra. psycopg2-binary is unchanged — the sync engine still uses psycopg2.

Why

Airflow recommends running PgBouncer in front of PostgreSQL in production. asyncpg uses named server-side prepared statements, which break under transaction-mode PgBouncer unless prepared-statement caching is explicitly disabled. psycopg3 is safe behind transaction-mode PgBouncer with zero configuration, so the default async engine now works out of the box in recommended production deployments. A safe default beats a documentation note operators miss.

asyncpg was originally chosen only because Airflow was pinned to SQLAlchemy 1.4; that constraint is gone (airflow-core requires sqlalchemy[asyncio]>=2.0.48), and psycopg3 serves both sync and async from a single driver.

Keeping asyncpg

asyncpg remains fully supported as a throughput opt-in:

pip install 'apache-airflow-providers-postgres[asyncpg]'

[database]
sql_alchemy_conn_async = postgresql+asyncpg://<user>:<password>@<host>/<db>

Behind transaction-mode PgBouncer, also disable asyncpg's prepared-statement caching via sql_alchemy_connect_args_async (a dict defined in airflow_local_settings.py):

# airflow_local_settings.py
connect_args_async = {"statement_cache_size": 0, "prepared_statement_cache_size": 0}

Notes for reviewers

• The derived async dialect is postgresql+psycopg_async:// (explicit form), not the postgresql+psycopg shorthand mentioned in the issue body — chosen deliberately for explicitness, matching the explicit-sync-driver direction of Make PostgreSQL SQLAlchemy driver explicit (postgresql+psycopg2://) #68314.
• Scope: SQLite (aiosqlite) and MySQL (aiomysql) async derivations are unchanged, and an explicitly configured sql_alchemy_conn_async is never rewritten.
• psycopg2 is intentionally kept as the sync hard dependency in this PR to keep it reviewable. Removing psycopg2 and migrating the sync engine to psycopg3 (aligned with the SQLAlchemy 2.1 upgrade, where the postgresql:// default flips to psycopg3) is tracked in Remove psycopg2 and migrate the sync Postgres driver to psycopg3 #68453.
• Merge ordering: depends on Make PostgreSQL SQLAlchemy driver explicit (postgresql+psycopg2://) #68314 (explicit sync postgresql+psycopg2://). Use async DB session for Execution API task-instance heartbeat #67800 (async heartbeat) will rebase its docs onto this change.
• Bulk-write caveat: psycopg3-async is markedly slower on batch INSERT/COPY in localhost benchmarks. Today's async routes are single-row OLTP, so this does not affect the switch, but bulk paths must be validated before they migrate to async (Migrate API endpoints from sync to async DB sessions #67799).

Validation

Validated end to end against a real transaction-mode PgBouncer in front of Postgres (dev/pgbouncer_e2e/): the default-derived psycopg_async engine runs repeated single-row and row-returning queries with no named prepared statements left on the backend, and the documented asyncpg opt-in recipe behaves the same.

Appendix: E2E tests

docker-compose.yaml

# Throwaway stack for validating the default async metadata engine behind
# transaction-mode PgBouncer (issue #67801 acceptance check). Not for production.
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    ports:
      - "55432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U airflow"]
      interval: 2s
      timeout: 2s
      retries: 30

  pgbouncer:
    image: edoburu/pgbouncer:latest
    environment:
      DB_HOST: postgres
      DB_USER: airflow
      DB_PASSWORD: airflow
      DB_NAME: airflow
      POOL_MODE: transaction
      AUTH_TYPE: scram-sha-256
      LISTEN_PORT: "6432"
    ports:
      - "56432:6432"
    depends_on:
      postgres:
        condition: service_healthy

pgbouncer_e2e.py

"""
E2E validation for issue #67801: the default-derived async Postgres engine must work
behind transaction-mode PgBouncer with no extra configuration, and the documented
asyncpg opt-in recipe must also work.

Run after `docker compose -f dev/pgbouncer_e2e/docker-compose.yaml up -d`:

    uv run --project airflow-core python dev/pgbouncer_e2e/run_e2e.py
"""

from __future__ import annotations

import asyncio
import os
import sys

PGBOUNCER_SYNC_URL = os.environ.get(
    "E2E_SYNC_URL", "postgresql+psycopg2://airflow:airflow@localhost:56432/airflow"
)
BACKEND_DIRECT_URL = os.environ.get(
    "E2E_BACKEND_URL", "postgresql+psycopg2://airflow:airflow@localhost:55432/airflow"
)
NUM_QUERIES = 50


def count_backend_prepared_statements(samples: int = 20) -> int:
    """
    Count named prepared statements visible on PgBouncer's pooled backend connections.

    ``pg_prepared_statements`` is connection-local, so the leftovers from a previous client
    are only visible when PgBouncer hands us the same backend connection. Sample many
    short transactions through PgBouncer (transaction mode rotates backends) and report
    the maximum seen on any backend.
    """
    from sqlalchemy import create_engine, text
    from sqlalchemy.pool import NullPool

    pgbouncer_url = PGBOUNCER_SYNC_URL
    worst = 0
    engine = create_engine(pgbouncer_url, poolclass=NullPool)
    for _ in range(samples):
        with engine.connect() as conn:
            count = conn.execute(text("SELECT count(*) FROM pg_prepared_statements")).scalar()
            worst = max(worst, int(count or 0))
    engine.dispose()
    return worst


async def exercise_async_engine(async_url: str, connect_args: dict) -> None:
    from sqlalchemy import text
    from sqlalchemy.ext.asyncio import create_async_engine

    engine = create_async_engine(async_url, connect_args=connect_args, pool_pre_ping=True)
    for i in range(NUM_QUERIES):
        async with engine.connect() as conn:
            value = (await conn.execute(text("SELECT 1"))).scalar()
            if value != 1:
                raise RuntimeError(f"query {i} returned {value!r}")
            ti = (
                await conn.execute(text("SELECT typname FROM pg_type WHERE typname = 'int4' LIMIT 1"))
            ).scalar()
            if ti != "int4":
                raise RuntimeError(f"row-returning query {i} returned {ti!r}")
    await engine.dispose()


def derive_default_async_url() -> str:
    os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = PGBOUNCER_SYNC_URL
    os.environ.pop("AIRFLOW__DATABASE__SQL_ALCHEMY_CONN_ASYNC", None)
    from airflow import settings

    settings.configure_vars()
    derived = settings.SQL_ALCHEMY_CONN_ASYNC
    print(f"derived async URL: {derived}")
    if not derived.startswith("postgresql+psycopg_async://"):
        raise RuntimeError(f"unexpected derived async URL: {derived}")
    return derived


def main() -> int:
    # Step 1+2: default derivation, then 50 queries through PgBouncer with the derived engine
    derived = derive_default_async_url()
    asyncio.run(exercise_async_engine(derived, connect_args={}))
    print(f"step 2 OK: {NUM_QUERIES} queries via default psycopg_async engine through PgBouncer")

    # Step 3: no named prepared statements may remain on the backend
    leftover = count_backend_prepared_statements()
    print(f"step 3: backend prepared statements after psycopg run: {leftover}")
    if leftover != 0:
        print("FAIL: psycopg run left prepared statements on the backend")
        return 1

    # Step 4: explicit asyncpg opt-in with the documented PgBouncer-safe recipe
    asyncpg_url = PGBOUNCER_SYNC_URL.replace("postgresql+psycopg2://", "postgresql+asyncpg://")
    asyncio.run(
        exercise_async_engine(
            asyncpg_url,
            connect_args={"statement_cache_size": 0, "prepared_statement_cache_size": 0},
        )
    )
    leftover = count_backend_prepared_statements()
    print(f"step 4: backend prepared statements after asyncpg recipe run: {leftover}")
    if leftover != 0:
        print("FAIL: asyncpg recipe run left prepared statements on the backend")
        return 1

    print("E2E PASS")
    return 0


if __name__ == "__main__":
    sys.exit(main())

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Code (Fable 5) following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Dev-iL]

Caveat: psycopg3 uses server-side prepared statements (via protocol-level PQsendPrepare/PQsendQueryPrepared), not client-side. Its default prepare_threshold=5 means it will create server-side prepared statements after 5 executions of the same query on a connection. Neither Airflow nor SQLAlchemy sets prepare_threshold=None.

The psycopg3 docs explicitly warn that poolers are not compatible with prepared statements unless:

• PgBouncer >= 1.22 (with max_prepared_statements > 0)
• libpq from PostgreSQL >= 17 (client-side)
• psycopg >= 3.2

Our Helm chart ships PgBouncer 1.23.1 and pins psycopg >= 3.2.9, so the first and third conditions are met for chart users. But CI tests Postgres 14–18 (default 14), so the libpq requirement (>= 17) is not always satisfied. Users running their own PgBouncer < 1.22 would also hit this.

Should we consider:

1. Adding PgBouncer to the integration test matrix (at least for the async engine path) to catch regressions across the version combinations?
2. Setting prepare_threshold=None as the safe default in settings.py for the async engine when no explicit connect_args_async is configured — making the "no extra configuration" claim actually hold for all versions?

[github-actions]

uv.lock on main just moved via #68856 ("Revert airflowctl dependency from airflow-core"), commit 6877aea and this PR currently conflicts.

Quickest fix:

git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease

Automated nudge — ignore if you're not ready to rebase. This comment is updated in place on future uv.lock bumps.

[Dev-iL]

@ashb @uranusjr what do you say?

Should we consider:

1. Adding PgBouncer to the integration test matrix (at least for the async engine path) to catch regressions across the version combinations?
2. Setting prepare_threshold=None as the safe default in settings.py for the async engine when no explicit connect_args_async is configured, making the "no extra configuration" claim actually hold for all versions?

[shahar1]

Overall LGTM, one comment to address from my prespective.
Also, could you please rebase onto latest main? (there seems to be a drift in some packages within uv.lock)

Edit: when the order of PR merges matters, it's important also to state it at the top of the description to avoid mistakes by committers :) Added it for you.

[uranusjr]

Would this change cause a breakage if core is upgraded without a provider upgrade? (i.e. use a new Airflow version without upgrading providers to a version that includes #69089)

[Dev-iL]

Great catch! Yes, this is indeed flawed.

Reproduced locally: airflow-core ships no Postgres driver, so with a new core and a pre-#69089 provider (which installs asyncpg but not psycopg), the derived postgresql+psycopg_async:// URL raises ModuleNotFoundError: No module named 'psycopg' at component startup.

I changed the approach to feature-detect via the existing _USE_PSYCOPG3 flag, which falls back to asyncpg when psycopg3 isn't installed. The database-setup docs and the newsfragment were updated accordingly.

The breakage of older core is documented in the provider changelog note (#69089): those deployments need the asyncpg extra or an explicitly configured sql_alchemy_conn_async.