This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f10ebb54222 Switch the default async Postgres driver from asyncpg to
psycopg3 (#68496)
f10ebb54222 is described below
commit f10ebb5422218d2be7a0a7062a9c7f94392473f2
Author: Dev-iL <[email protected]>
AuthorDate: Sat Jul 4 21:32:18 2026 +0300
Switch the default async Postgres driver from asyncpg to psycopg3 (#68496)
The derived async metadata-DB URL now prefers postgresql+psycopg_async://,
which is safe behind transaction-mode PgBouncer with no configuration, and
falls back to asyncpg when psycopg3 is not installed — so a newer core keeps
working with an older postgres provider that ships asyncpg only.
The matching driver ships in the postgres provider (>=7.0.0), which keeps
asyncpg available via an opt-in extra and an explicit sql_alchemy_conn_async
URL. This default takes effect in Airflow 3.4.0.
---
airflow-core/docs/howto/set-up-database.rst | 50 ++++++++++++++++++++----
airflow-core/newsfragments/68496.significant.rst | 32 +++++++++++++++
airflow-core/src/airflow/settings.py | 8 +++-
airflow-core/tests/unit/core/test_settings.py | 33 ++++++++++++++++
4 files changed, 113 insertions(+), 10 deletions(-)
diff --git a/airflow-core/docs/howto/set-up-database.rst
b/airflow-core/docs/howto/set-up-database.rst
index 43cafdb22c4..ec9604f2ccb 100644
--- a/airflow-core/docs/howto/set-up-database.rst
+++ b/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.
diff --git a/airflow-core/newsfragments/68496.significant.rst
b/airflow-core/newsfragments/68496.significant.rst
new file mode 100644
index 00000000000..5e0d3225764
--- /dev/null
+++ b/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 [...]
+
+**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
diff --git a/airflow-core/src/airflow/settings.py
b/airflow-core/src/airflow/settings.py
index 56fc3d07ced..54150dd8bfd 100644
--- a/airflow-core/src/airflow/settings.py
+++ b/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]
diff --git a/airflow-core/tests/unit/core/test_settings.py
b/airflow-core/tests/unit/core/test_settings.py
index 588567181c5..c703ee05ab9 100644
--- a/airflow-core/tests/unit/core/test_settings.py
+++ b/airflow-core/tests/unit/core/test_settings.py
@@ -434,6 +434,39 @@ def test_sqlite_relative_path(value, expectation):
settings.configure_orm()
[email protected](
+ ("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."""
