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

airflow-core/docs/howto/set-up-database.rst

@@ -184,9 +184,9 @@ in the Postgres documentation to learn more.
 
 .. warning::
 
-   When you use SQLAlchemy 1.4.0+, you need to use ``postgresql+psycopg2://`` as the database in the ``sql_alchemy_conn``.
-   In the previous versions of SQLAlchemy it was possible to use ``postgres://``, but using it in
-   SQLAlchemy 1.4.0+ results in:
+   On SQLAlchemy 1.4+, use ``postgresql://`` as the connection string schema in ``sql_alchemy_conn``.
+   In earlier versions of SQLAlchemy it was possible to use ``postgres://``, but doing so in
+   SQLAlchemy 1.4+ results in:
 
    .. code-block::
 
@@ -195,16 +195,50 @@ in the Postgres documentation to learn more.
               )
       E       sqlalchemy.exc.NoSuchModuleError: Can't load plugin: sqlalchemy.dialects:postgres
 
-   If you cannot change the prefix of your URL immediately, Airflow continues to work with SQLAlchemy
-   1.3 and you can downgrade SQLAlchemy, but we recommend to update the prefix.
-
    Details in the `SQLAlchemy Changelog <https://docs.sqlalchemy.org/en/20/changelog/changelog_14.html#change-3687655465c25a39b968b4f5f6e9170b>`_.
 
-We recommend using the ``psycopg2`` driver and specifying it in your SqlAlchemy connection string.
+We recommend specifying a driver, such as ``psycopg2``, explicitly in your SQLAlchemy connection string:
+
+.. code-block:: text
+
+   postgresql+<driver>://<user>:<password>@<host>/<db>
+
+Async SQLAlchemy engine and the async driver
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+In addition to the synchronous engine, Airflow maintains an async SQLAlchemy engine for the metadata database (used e.g. by async API endpoints).
+When ``[database] sql_alchemy_conn_async`` is not set, its URL is derived from ``sql_alchemy_conn`` using ``psycopg`` (psycopg3) as the async driver:
 
 .. code-block:: text
 
-   postgresql+psycopg2://<user>:<password>@<host>/<db>
+   postgresql+psycopg_async://<user>:<password>@<host>/<db>
+
+psycopg3 is the default because it is safe behind transaction-mode PgBouncer (recommended for all production Postgres installations, see the note below) with no extra configuration.
+psycopg3 `defers statement preparation <https://www.psycopg.org/psycopg3/docs/advanced/prepare.html>`_ behind a threshold (default 5 executions) and lets you disable it entirely with ``prepare_threshold=None``.
+If psycopg3 is not installed — for example with an older ``apache-airflow-providers-postgres`` release that ships ``asyncpg`` but not ``psycopg`` — Airflow falls back to deriving an ``asyncpg`` URL instead.
+
+If you need the extra throughput of ``asyncpg``, opt in by installing the ``apache-airflow-providers-postgres[asyncpg]`` extra and setting the async URL explicitly:
+
+.. code-block:: ini
+
+   [database]
+   sql_alchemy_conn_async = postgresql+asyncpg://<user>:<password>@<host>/<db>
+
+asyncpg uses `named server-side prepared statements <https://magicstack.github.io/asyncpg/current/faq.html#why-am-i-getting-prepared-statement-errors>`_, which break under transaction-mode PgBouncer.
+If you run asyncpg behind transaction-mode PgBouncer, you must disable its prepared-statement caching by pointing ``sql_alchemy_connect_args_async`` at a dictionary defined in your ``airflow_local_settings.py``:
+
+.. code-block:: python
+
+   # airflow_local_settings.py
+   connect_args_async = {
+       "statement_cache_size": 0,
+       "prepared_statement_cache_size": 0,
+   }
+
+.. code-block:: ini
+
+   [database]
+   sql_alchemy_connect_args_async = airflow_local_settings.connect_args_async
 
 Also note that since SqlAlchemy does not expose a way to target a specific schema in the database URI, you need to ensure schema ``public`` is in your Postgres user's search_path.
 

airflow-core/newsfragments/68496.significant.rst

@@ -0,0 +1,32 @@
+Default async Postgres driver changed from asyncpg to psycopg3
+
+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://``) when psycopg3 is installed. If psycopg3 is not installed — for example with an older ``apache-airflow-providers-postgres`` release that ships ``asyncpg`` but not ``psycopg`` — the derived URL keeps using asyncpg, so upgrading core without upgrading the provider does not break the async engine.
+
+**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 no extra configuration, so the default async engine now works out of the box in recommended production deployments.
+
+The SQLite (``aiosqlite``) and MySQL (``aiomysql``) async URL derivations are unchanged, and an explicitly configured ``sql_alchemy_conn_async`` is never rewritten.
+
+**Packaging:** the ``apache-airflow-providers-postgres`` distribution now installs ``psycopg`` (psycopg3) by default, and ``asyncpg`` is no longer installed by default. Instead, it moved to the ``asyncpg`` optional extra. Note that Airflow versions older than 3.4.0 always derive an asyncpg URL (they have no psycopg3 fallback), so running the new provider release on an older core requires either the ``asyncpg`` extra or an explicitly configured ``sql_alchemy_conn_async``.
+
+**To keep using asyncpg**, install the extra and configure the async URL explicitly:
+
+.. code-block:: bash
+
+    pip install 'apache-airflow-providers-postgres[asyncpg]'
+
+.. code-block:: ini
+
+    [database]
+    sql_alchemy_conn_async = postgresql+asyncpg://<user>:<password>@<host>/<db>
+
+When running behind transaction-mode PgBouncer, also disable asyncpg's prepared-statement caching by pointing ``sql_alchemy_connect_args_async`` at a dict defined in ``airflow_local_settings.py``:
+
+.. code-block:: python
+
+    # airflow_local_settings.py
+    connect_args_async = {"statement_cache_size": 0, "prepared_statement_cache_size": 0}
+
+.. code-block:: ini
+
+    [database]
+    sql_alchemy_connect_args_async = airflow_local_settings.connect_args_async

airflow-core/src/airflow/settings.py

@@ -241,8 +241,12 @@ def load_policy_plugins(pm: pluggy.PluginManager):
 
 
 def _get_async_conn_uri_from_sync(sync_uri):
-    AIO_LIBS_MAPPING = {"sqlite": "aiosqlite", "postgresql": "asyncpg", "mysql": "aiomysql"}
-    """Mapping of sync scheme to async scheme."""
+    # Mapping of backend to async driver:
+    AIO_LIBS_MAPPING = {
+        "sqlite": "aiosqlite",
+        "postgresql": "psycopg_async" if _USE_PSYCOPG3 else "asyncpg",
+        "mysql": "aiomysql",
+    }
 
     scheme, rest = sync_uri.split(":", maxsplit=1)
     scheme = scheme.split("+", maxsplit=1)[0]

airflow-core/tests/unit/core/test_settings.py

@@ -434,6 +434,39 @@ def test_sqlite_relative_path(value, expectation):
             settings.configure_orm()
 
 
+@pytest.mark.parametrize(
+    ("sync_uri", "use_psycopg3", "expected_async_uri"),
+    [
+        ("postgresql://user:pass@host/db", True, "postgresql+psycopg_async://user:pass@host/db"),
+        ("postgresql://user:pass@host/db", False, "postgresql+asyncpg://user:pass@host/db"),
+        ("postgresql+psycopg2://user:pass@host/db", True, "postgresql+psycopg_async://user:pass@host/db"),
+        ("postgresql+psycopg2://user:pass@host/db", False, "postgresql+asyncpg://user:pass@host/db"),
+        ("sqlite:////root/airflow.db", True, "sqlite+aiosqlite:////root/airflow.db"),
+        ("mysql://user:pass@host/db", True, "mysql+aiomysql://user:pass@host/db"),
+        ("mssql://user:pass@host/db", True, "mssql://user:pass@host/db"),
+    ],
+)
+def test_get_async_conn_uri_from_sync(sync_uri, use_psycopg3, expected_async_uri):
+    from airflow import settings
+
+    with patch.object(settings, "_USE_PSYCOPG3", use_psycopg3):
+        assert settings._get_async_conn_uri_from_sync(sync_uri) == expected_async_uri
+
+
+def test_explicit_sql_alchemy_conn_async_is_not_rewritten():
+    from airflow import settings
+
+    from tests_common.test_utils.config import conf_vars
+
+    explicit_async_uri = "postgresql+asyncpg://user:pass@host/db"
+    try:
+        with conf_vars({("database", "sql_alchemy_conn_async"): explicit_async_uri}):
+            settings.configure_vars()
+            assert explicit_async_uri == settings.SQL_ALCHEMY_CONN_ASYNC
+    finally:
+        settings.configure_vars()
+
+
 class TestDisposeOrm:
     """Tests for dispose_orm() async engine disposal."""