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 45a93a1e1fb Make psycopg (v3) the default synchronous Postgres driver
in airflow-core (#69469)
45a93a1e1fb is described below
commit 45a93a1e1fbadcdf46648fe7658b61ea55fff9d3
Author: Dev-iL <[email protected]>
AuthorDate: Wed Jul 22 21:08:47 2026 +0300
Make psycopg (v3) the default synchronous Postgres driver in airflow-core
(#69469)
---
.../logging-monitoring/logging-tasks.rst | 2 +-
.../modules_management.rst | 2 +-
airflow-core/docs/core-concepts/multi-team.rst | 8 ++--
.../docs/howto/docker-compose/docker-compose.yaml | 4 +-
airflow-core/docs/howto/set-up-database.rst | 26 +++++++++++-
airflow-core/newsfragments/69469.improvement.rst | 1 +
airflow-core/src/airflow/configuration.py | 12 ++++--
...nconsistency_between_ORM_and_migration_files.py | 2 +-
...0_replace_deadline_inline_callback_with_fkey.py | 6 +--
airflow-core/src/airflow/models/serialized_dag.py | 6 ++-
airflow-core/src/airflow/settings.py | 7 ++--
.../unit/api_fastapi/common/test_exceptions.py | 20 ++++++++-
airflow-core/tests/unit/core/test_configuration.py | 38 +++++++++++++++--
.../tests/unit/core/test_sqlalchemy_config.py | 24 +++++++++++
airflow-e2e-tests/docker/openlineage-compat-db.yml | 48 ++++++++++++++++++++++
.../tests/airflow_e2e_tests/conftest.py | 9 +++-
.../tests/airflow_e2e_tests/constants.py | 5 +++
chart/values.yaml | 6 ++-
scripts/ci/docker-compose/backend-postgres.yml | 4 +-
task-sdk-integration-tests/docker-compose.yaml | 2 +-
20 files changed, 198 insertions(+), 34 deletions(-)
diff --git
a/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst
b/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst
index 3f2a70e49c9..e947fb5c9d8 100644
---
a/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst
+++
b/airflow-core/docs/administration-and-deployment/logging-monitoring/logging-tasks.rst
@@ -150,7 +150,7 @@ the example below.
version | 2.9.0.dev0
executor | LocalExecutor
task_logging_handler |
airflow.utils.log.file_task_handler.FileTaskHandler
- sql_alchemy_conn |
postgresql+psycopg2://postgres:airflow@postgres/airflow
+ sql_alchemy_conn |
postgresql+psycopg://postgres:airflow@postgres/airflow
dags_folder | /files/dags
plugins_folder | /root/airflow/plugins
base_log_folder | /root/airflow/logs
diff --git
a/airflow-core/docs/administration-and-deployment/modules_management.rst
b/airflow-core/docs/administration-and-deployment/modules_management.rst
index 4ea73ea6fd7..62f3430da22 100644
--- a/airflow-core/docs/administration-and-deployment/modules_management.rst
+++ b/airflow-core/docs/administration-and-deployment/modules_management.rst
@@ -293,7 +293,7 @@ Below is the sample output of the ``airflow info`` command:
Config info
executor | LocalExecutor
task_logging_handler | airflow.utils.log.file_task_handler.FileTaskHandler
- sql_alchemy_conn |
postgresql+psycopg2://postgres:airflow@postgres/airflow
+ sql_alchemy_conn |
postgresql+psycopg://postgres:airflow@postgres/airflow
dags_folder | /files/dags
plugins_folder | /root/airflow/plugins
base_log_folder | /root/airflow/logs
diff --git a/airflow-core/docs/core-concepts/multi-team.rst
b/airflow-core/docs/core-concepts/multi-team.rst
index 6a9d8a2ec63..b45b0379b7f 100644
--- a/airflow-core/docs/core-concepts/multi-team.rst
+++ b/airflow-core/docs/core-concepts/multi-team.rst
@@ -486,7 +486,7 @@ is uppercase.
export AIRFLOW__TEAM_B___CELERY__BROKER_URL="redis://team-b-redis:6379/0"
# team_b's Celery result backend
- export
AIRFLOW__TEAM_B___CELERY__RESULT_BACKEND="db+postgresql+psycopg2://team-b-db/celery_results"
+ export
AIRFLOW__TEAM_B___CELERY__RESULT_BACKEND="db+postgresql+psycopg://team-b-db/celery_results"
Via Config File
"""""""""""""""
@@ -499,17 +499,17 @@ name followed by an equals sign:
# Global celery settings (used by the global executor, NOT as a fallback
for teams)
[celery]
broker_url = redis://default-redis:6379/0
- result_backend = db+postgresql+psycopg2://default-db/celery_results
+ result_backend = db+postgresql+psycopg://default-db/celery_results
# team_a overrides
[team_a=celery]
broker_url = redis://team-a-redis:6379/0
- result_backend = db+postgresql+psycopg2://team-a-db/celery_results
+ result_backend = db+postgresql+psycopg://team-a-db/celery_results
# team_b overrides
[team_b=celery]
broker_url = redis://team-b-redis:6379/0
- result_backend = db+postgresql+psycopg2://team-b-db/celery_results
+ result_backend = db+postgresql+psycopg://team-b-db/celery_results
Dag Bundle to Team Association
------------------------------
diff --git a/airflow-core/docs/howto/docker-compose/docker-compose.yaml
b/airflow-core/docs/howto/docker-compose/docker-compose.yaml
index 6a8891f3f58..b68f1afd3cc 100644
--- a/airflow-core/docs/howto/docker-compose/docker-compose.yaml
+++ b/airflow-core/docs/howto/docker-compose/docker-compose.yaml
@@ -57,8 +57,8 @@ x-airflow-common:
&airflow-common-env
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__CORE__AUTH_MANAGER:
airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:
postgresql+psycopg2://airflow:airflow@postgres/airflow
- AIRFLOW__CELERY__RESULT_BACKEND:
db+postgresql+psycopg2://airflow:airflow@postgres/airflow
+ AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:
postgresql+psycopg://airflow:airflow@postgres/airflow
+ AIRFLOW__CELERY__RESULT_BACKEND:
db+postgresql+psycopg://airflow:airflow@postgres/airflow
AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY}
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
diff --git a/airflow-core/docs/howto/set-up-database.rst
b/airflow-core/docs/howto/set-up-database.rst
index ec9604f2ccb..03e0a7a627d 100644
--- a/airflow-core/docs/howto/set-up-database.rst
+++ b/airflow-core/docs/howto/set-up-database.rst
@@ -197,12 +197,34 @@ in the Postgres documentation to learn more.
Details in the `SQLAlchemy Changelog
<https://docs.sqlalchemy.org/en/20/changelog/changelog_14.html#change-3687655465c25a39b968b4f5f6e9170b>`_.
-We recommend specifying a driver, such as ``psycopg2``, explicitly in your
SQLAlchemy connection string:
+We recommend specifying a driver, such as ``psycopg``, explicitly in your
SQLAlchemy connection string:
.. code-block:: text
postgresql+<driver>://<user>:<password>@<host>/<db>
+Sync SQLAlchemy engine and the sync driver
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+``psycopg`` (psycopg3) is the default synchronous driver for the metadata
database. When ``sql_alchemy_conn`` omits a driver
+(a bare ``postgresql://`` scheme) or still uses the legacy ``postgres://`` /
``postgres+psycopg2://`` schemes, Airflow
+rewrites it to:
+
+.. code-block:: text
+
+ postgresql+psycopg://<user>:<password>@<host>/<db>
+
+If you need the old ``psycopg2`` driver, install the
``apache-airflow-providers-postgres[psycopg2]`` extra and set the
+connection string explicitly:
+
+.. code-block:: ini
+
+ [database]
+ sql_alchemy_conn = postgresql+psycopg2://<user>:<password>@<host>/<db>
+
+An explicit ``postgresql+psycopg2://`` URL is never rewritten. If ``psycopg2``
is not installed while such a URL is
+configured, Airflow fails loudly at connection time rather than silently
falling back to a different driver.
+
Async SQLAlchemy engine and the async driver
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -387,7 +409,7 @@ For instance, you can specify a database schema where
Airflow will create its re
.. code-block:: bash
- export
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://postgres@localhost:5432/my_database?options=-csearch_path%3Dairflow"
+ export
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg://postgres@localhost:5432/my_database?options=-csearch_path%3Dairflow"
export AIRFLOW__DATABASE__SQL_ALCHEMY_SCHEMA="airflow"
Note the ``search_path`` at the end of the ``SQL_ALCHEMY_CONN`` database URL.
diff --git a/airflow-core/newsfragments/69469.improvement.rst
b/airflow-core/newsfragments/69469.improvement.rst
new file mode 100644
index 00000000000..2260dc1287d
--- /dev/null
+++ b/airflow-core/newsfragments/69469.improvement.rst
@@ -0,0 +1 @@
+``psycopg`` (psycopg3) is now the default synchronous Postgres driver for the
metadata database, mirroring the existing async default; a bare
``postgresql://`` or legacy ``postgres://``/``postgres+psycopg2://``
``sql_alchemy_conn`` is now rewritten to ``postgresql+psycopg://`` instead of
``postgresql+psycopg2://``, while an explicit ``postgresql+psycopg2://``
connection string is left untouched (install the
``apache-airflow-providers-postgres[psycopg2]`` extra to keep using psycopg2).
diff --git a/airflow-core/src/airflow/configuration.py
b/airflow-core/src/airflow/configuration.py
index d43eb715d32..4eb92054cea 100644
--- a/airflow-core/src/airflow/configuration.py
+++ b/airflow-core/src/airflow/configuration.py
@@ -28,6 +28,7 @@ import warnings
from base64 import b64encode
from collections.abc import Callable
from configparser import ConfigParser
+from importlib.util import find_spec
from inspect import ismodule
from io import StringIO
from re import Pattern
@@ -372,13 +373,18 @@ class AirflowConfigParser(_SharedAirflowConfigParser):
Upgrade SQL schemas.
As of SQLAlchemy 1.4, schemes `postgres+psycopg2` and `postgres`
- must be replaced with `postgresql+psycopg2`. The bare `postgresql`
- scheme is also upgraded to make the psycopg2 driver explicit.
+ must be replaced with `postgresql+psycopg` if the psycopg (v3) driver
+ is installed, or `postgresql+psycopg2` otherwise. The bare `postgresql`
+ scheme is upgraded the same way to make the driver explicit.
"""
section, key = "database", "sql_alchemy_conn"
old_value = self.get(section, key, _extra_stacklevel=1)
bad_schemes = ["postgres+psycopg2", "postgres", "postgresql"]
- good_scheme = "postgresql+psycopg2"
+ # The provider-side hooks (common.sql / postgres / amazon) also gate
psycopg (v3) on an
+ # ``_is_sqlalchemy_2()`` check, because they support Airflow 2.11 on
SQLAlchemy 1.4, which has
+ # no native ``postgresql+psycopg`` dialect. airflow-core pins
``sqlalchemy>=2.0``, so that
+ # dialect is always present when the package is importable —
``find_spec`` alone suffices here.
+ good_scheme = "postgresql+psycopg" if find_spec("psycopg") is not None
else "postgresql+psycopg2"
parsed = urlsplit(old_value)
if parsed.scheme in bad_schemes:
warnings.warn(
diff --git
a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py
b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py
index fa24916df6f..64531a957b3 100644
---
a/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py
+++
b/airflow-core/src/airflow/migrations/versions/0017_2_9_2_fix_inconsistency_between_ORM_and_migration_files.py
@@ -110,7 +110,7 @@ def upgrade():
batch_op.create_unique_constraint(batch_op.f("connection_conn_id_uq"),
["conn_id"])
max_cons = sa.table("dag", sa.column("max_consecutive_failed_dag_runs"))
-
op.execute(max_cons.update().values(max_consecutive_failed_dag_runs=literal("0")))
+
op.execute(max_cons.update().values(max_consecutive_failed_dag_runs=literal(0)))
with op.batch_alter_table("dag") as batch_op:
batch_op.alter_column("max_consecutive_failed_dag_runs",
existing_type=sa.Integer(), nullable=False)
diff --git
a/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py
b/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py
index 0cab3d57c25..bf1f6651092 100644
---
a/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py
+++
b/airflow-core/src/airflow/migrations/versions/0094_3_2_0_replace_deadline_inline_callback_with_fkey.py
@@ -92,7 +92,7 @@ def _upgrade_postgresql(conn, batch_size):
'__var', json_build_object(
'path', b.cb_path,
'kwargs', b.cb_kwargs,
- 'prefix', :prefix,
+ 'prefix', CAST(:prefix AS text),
'dag_id', b.dag_id
),
'__type', 'dict'
@@ -325,7 +325,7 @@ def _downgrade_postgresql(conn, batch_size):
'path', c.data::jsonb->'__var'->>'path',
'kwargs', c.data::jsonb->'__var'->'kwargs'
),
- '__classname__', :classname,
+ '__classname__', CAST(:classname AS text),
'__version__', 0
)::json,
callback_state = CASE
@@ -360,7 +360,7 @@ def _downgrade_postgresql(conn, batch_size):
'path', c.data::jsonb->'__var'->>'path',
'kwargs', c.data::jsonb->'__var'->'kwargs'
),
- '__classname__', :classname,
+ '__classname__', CAST(:classname AS text),
'__version__', 0
)::json,
callback_state = CASE
diff --git a/airflow-core/src/airflow/models/serialized_dag.py
b/airflow-core/src/airflow/models/serialized_dag.py
index fbff450a993..9dda74d42b5 100644
--- a/airflow-core/src/airflow/models/serialized_dag.py
+++ b/airflow-core/src/airflow/models/serialized_dag.py
@@ -28,7 +28,7 @@ from uuid import UUID
import uuid6
from sqlalchemy import JSON, ForeignKey, LargeBinary, String, Uuid, exists,
select, tuple_, update
-from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import ARRAY, JSONB
from sqlalchemy.orm import Mapped, backref, foreign, mapped_column,
relationship
from sqlalchemy.sql.expression import func, literal
@@ -955,7 +955,9 @@ class SerializedDagModel(Base):
elif dialect == "postgresql":
# Use #> operator which works for both JSON and JSONB types
# Returns the JSON sub-object at the specified path
- data_col_to_select =
cls._data.op("#>")(literal('{"dag","dag_dependencies"}'))
+ data_col_to_select = cls._data.op("#>")(
+ literal(["dag", "dag_dependencies"], type_=ARRAY(String))
+ )
load_json = lambda x: x
else:
data_col_to_select = func.json_extract_path(cls._data, "dag",
"dag_dependencies")
diff --git a/airflow-core/src/airflow/settings.py
b/airflow-core/src/airflow/settings.py
index 54150dd8bfd..0a0e1c66197 100644
--- a/airflow-core/src/airflow/settings.py
+++ b/airflow-core/src/airflow/settings.py
@@ -524,15 +524,16 @@ def _is_sqlite_in_memory(url: str) -> bool:
def prepare_engine_args(disable_connection_pool=False, pool_class=None):
"""Prepare SQLAlchemy engine args."""
+ use_psycopg2_tuning = SQL_ALCHEMY_CONN.startswith("postgresql+psycopg2")
DEFAULT_ENGINE_ARGS: dict[str, dict[str, Any]] = {
"postgresql": (
{
"insertmanyvalues_page_size": 10000,
}
| (
- {}
- if _USE_PSYCOPG3
- else {"executemany_mode": "values_plus_batch",
"executemany_batch_page_size": 2000}
+ {"executemany_mode": "values_plus_batch",
"executemany_batch_page_size": 2000}
+ if use_psycopg2_tuning
+ else {}
)
)
}
diff --git a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
index c38cbabb5e1..def8a0abd30 100644
--- a/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
+++ b/airflow-core/tests/unit/api_fastapi/common/test_exceptions.py
@@ -277,7 +277,22 @@ class TestUniqueConstraintErrorHandler:
exeinfo_response_error.value.detail.pop("message", None) # type:
ignore[attr-defined]
assert exeinfo_response_error.value.status_code ==
expected_exception.status_code
- assert exeinfo_response_error.value.detail == expected_exception.detail
+
+ # The exact rendered SQL (e.g. explicit driver-added type casts) and
trailing whitespace in
+ # the driver's own error text are driver-specific rendering details,
not meaningful content —
+ # match on the statement's target table and normalize trailing
newlines instead of an exact
+ # string compare.
+ response_detail = exeinfo_response_error.value.detail
+ expected_detail = expected_exception.detail
+ actual_statement = response_detail.pop("statement", None) # type:
ignore[attr-defined]
+ expected_statement = expected_detail.pop("statement", None)
+ response_detail["orig_error"] =
response_detail["orig_error"].rstrip("\n") # type: ignore[index]
+ expected_detail["orig_error"] =
expected_detail["orig_error"].rstrip("\n")
+
+ assert response_detail == expected_detail
+ assert expected_statement is not None
+ assert actual_statement is not None
+ assert f"INSERT INTO {expected_statement.split()[2]}" in
actual_statement
@patch("airflow.api_fastapi.common.exceptions.get_random_string",
return_value=MOCKED_ID)
@conf_vars({("api", "expose_stacktrace"): "False"})
@@ -391,6 +406,9 @@ class TestUniqueConstraintErrorHandler:
# Removes the stacktrace from response to remove during comparison.
response_detail.pop("message", None) # type: ignore[attr-defined]
+ # Trailing whitespace in the driver's own error text is a
driver-specific rendering detail.
+ response_detail["orig_error"] =
response_detail["orig_error"].rstrip("\n") # type: ignore[index]
+ expected_detail["orig_error"] =
expected_detail["orig_error"].rstrip("\n")
assert response_detail == expected_detail
assert "INSERT INTO dag_run" in actual_statement
assert exeinfo_response_error.value.detail == expected_exception.detail
diff --git a/airflow-core/tests/unit/core/test_configuration.py
b/airflow-core/tests/unit/core/test_configuration.py
index fc44c398e8e..11d1809fe9f 100644
--- a/airflow-core/tests/unit/core/test_configuration.py
+++ b/airflow-core/tests/unit/core/test_configuration.py
@@ -1124,16 +1124,16 @@ class TestConf:
("input_scheme", "expected_scheme"),
[
pytest.param(
- "postgres://user:pass@host/db",
"postgresql+psycopg2://user:pass@host/db", id="postgres"
+ "postgres://user:pass@host/db",
"postgresql+psycopg://user:pass@host/db", id="postgres"
),
pytest.param(
"postgres+psycopg2://user:pass@host/db",
- "postgresql+psycopg2://user:pass@host/db",
+ "postgresql+psycopg://user:pass@host/db",
id="postgres+psycopg2",
),
pytest.param(
"postgresql://user:pass@host/db",
- "postgresql+psycopg2://user:pass@host/db",
+ "postgresql+psycopg://user:pass@host/db",
id="postgresql-bare",
),
pytest.param(
@@ -1145,7 +1145,37 @@ class TestConf:
],
)
@mock.patch.dict("os.environ", {}, clear=False)
- def test_upgrade_postgres_metastore_conn(self, input_scheme,
expected_scheme):
+ @mock.patch("airflow.configuration.find_spec", return_value=object())
+ def test_upgrade_postgres_metastore_conn(self, mock_find_spec,
input_scheme, expected_scheme):
+ """Assumes psycopg (v3) is installed; see the sibling
``_without_psycopg3`` test for the
+ fallback, since whether psycopg3 is actually present varies across CI
environments
+ (e.g. a lowest-dependencies job that doesn't install it)."""
+ os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = input_scheme
+ test_conf = AirflowConfigParser()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", FutureWarning)
+ test_conf._upgrade_postgres_metastore_conn()
+ assert test_conf.get("database", "sql_alchemy_conn") == expected_scheme
+
+ @pytest.mark.parametrize(
+ ("input_scheme", "expected_scheme"),
+ [
+ pytest.param(
+ "postgres://user:pass@host/db",
"postgresql+psycopg2://user:pass@host/db", id="postgres"
+ ),
+ pytest.param(
+ "postgresql+psycopg2://user:pass@host/db",
+ "postgresql+psycopg2://user:pass@host/db",
+ id="postgresql+psycopg2-noop",
+ ),
+ ],
+ )
+ @mock.patch.dict("os.environ", {}, clear=False)
+ @mock.patch("airflow.configuration.find_spec", return_value=None)
+ def test_upgrade_postgres_metastore_conn_without_psycopg3(
+ self, mock_find_spec, input_scheme, expected_scheme
+ ):
+ """Falls back to psycopg2 when the psycopg (v3) package isn't
installed."""
os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = input_scheme
test_conf = AirflowConfigParser()
with warnings.catch_warnings():
diff --git a/airflow-core/tests/unit/core/test_sqlalchemy_config.py
b/airflow-core/tests/unit/core/test_sqlalchemy_config.py
index db211bab2c5..ab364dd9a7b 100644
--- a/airflow-core/tests/unit/core/test_sqlalchemy_config.py
+++ b/airflow-core/tests/unit/core/test_sqlalchemy_config.py
@@ -121,6 +121,30 @@ class TestSqlAlchemySettings:
assert "pool_recycle" not in engine_args
assert "pool_pre_ping" not in engine_args
+ @pytest.mark.parametrize(
+ ("conn_str", "use_psycopg3", "expect_psycopg2_tuning"),
+ [
+ ("postgresql+psycopg2://user:pass@host/db", True, True),
+ ("postgresql+psycopg2://user:pass@host/db", False, True),
+ ("postgresql+psycopg://user:pass@host/db", True, False),
+ ("postgresql+psycopg://user:pass@host/db", False, False),
+ ],
+ )
+ def test_prepare_engine_args_psycopg2_tuning_follows_configured_scheme(
+ self, conn_str, use_psycopg3, expect_psycopg2_tuning, monkeypatch
+ ):
+ """executemany tuning must follow the configured scheme, not whether
psycopg happens to be importable."""
+ monkeypatch.setattr(settings, "SQL_ALCHEMY_CONN", conn_str)
+ monkeypatch.setattr(settings, "_USE_PSYCOPG3", use_psycopg3)
+ engine_args = settings.prepare_engine_args()
+ assert engine_args["insertmanyvalues_page_size"] == 10000
+ if expect_psycopg2_tuning:
+ assert engine_args["executemany_mode"] == "values_plus_batch"
+ assert engine_args["executemany_batch_page_size"] == 2000
+ else:
+ assert "executemany_mode" not in engine_args
+ assert "executemany_batch_page_size" not in engine_args
+
@patch("airflow.settings.setup_event_handlers")
@patch("airflow.settings.scoped_session")
@patch("airflow.settings.sessionmaker")
diff --git a/airflow-e2e-tests/docker/openlineage-compat-db.yml
b/airflow-e2e-tests/docker/openlineage-compat-db.yml
new file mode 100644
index 00000000000..01c9e68e2fd
--- /dev/null
+++ b/airflow-e2e-tests/docker/openlineage-compat-db.yml
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Docker Compose override for the openlineage E2E compat matrix
(``--airflow-version``).
+#
+# docker-compose.yaml points sql_alchemy_conn/result_backend at the
``psycopg`` (v3) driver, matching
+# the current default. The compat matrix runs unmodified released
``apache/airflow:<version>`` images
+# for versions that only ever bundled psycopg2 by default (no psycopg3
provider dependency existed
+# yet), and does not reinstall the postgres provider on top of them — so the
psycopg3 dialect can't
+# load there. Pin the explicit psycopg2 scheme these images actually support.
+---
+x-airflow-common-compat-db-env: &airflow-common-compat-db-env
+ AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:
postgresql+psycopg2://airflow:airflow@postgres/airflow
+ AIRFLOW__CELERY__RESULT_BACKEND:
db+postgresql+psycopg2://airflow:airflow@postgres/airflow
+
+services:
+ airflow-apiserver:
+ environment:
+ <<: *airflow-common-compat-db-env
+ airflow-scheduler:
+ environment:
+ <<: *airflow-common-compat-db-env
+ airflow-dag-processor:
+ environment:
+ <<: *airflow-common-compat-db-env
+ airflow-triggerer:
+ environment:
+ <<: *airflow-common-compat-db-env
+ airflow-worker:
+ environment:
+ <<: *airflow-common-compat-db-env
+ airflow-init:
+ environment:
+ <<: *airflow-common-compat-db-env
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
index b97add595fd..5b45974f053 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/conftest.py
@@ -56,6 +56,7 @@ from airflow_e2e_tests.constants import (
LOCALSTACK_PATH,
LOGS_FOLDER,
NODE_IMAGE,
+ OPENLINEAGE_COMPAT_DB_COMPOSE_PATH,
OPENLINEAGE_COMPOSE_PATH,
OPENSEARCH_PATH,
PROVIDERS_MOUNT_CONTAINER_PATH,
@@ -630,7 +631,7 @@ def _setup_go_sdk_integration(dot_env_file, tmp_dir):
os.environ["ENV_FILE_PATH"] = str(dot_env_file)
-def _setup_openlineage_integration(dot_env_file, tmp_dir):
+def _setup_openlineage_integration(dot_env_file, tmp_dir, compose_file_names):
"""Set up the openlineage E2E test mode.
The OpenLineage system-test DAGs are the single source of truth;
``prepare_dags`` copies them
@@ -644,6 +645,10 @@ def _setup_openlineage_integration(dot_env_file, tmp_dir):
prepare_dags(tmp_dir / "dags")
copyfile(OPENLINEAGE_COMPOSE_PATH, tmp_dir / "openlineage.yml")
+ if os.environ.get("E2E_TARGET_AIRFLOW_VERSION", "").strip():
+ copyfile(OPENLINEAGE_COMPAT_DB_COMPOSE_PATH, tmp_dir /
"openlineage-compat-db.yml")
+ compose_file_names.append("openlineage-compat-db.yml")
+
def _build_ts_sdk_example_bundle(*, native=False):
build_commands = (
@@ -784,7 +789,7 @@ def spin_up_airflow_environment(tmp_path_factory:
pytest.TempPathFactory):
_setup_go_sdk_integration(dot_env_file, tmp_dir)
elif E2E_TEST_MODE == "openlineage":
compose_file_names.append("openlineage.yml")
- _setup_openlineage_integration(dot_env_file, tmp_dir)
+ _setup_openlineage_integration(dot_env_file, tmp_dir,
compose_file_names)
elif E2E_TEST_MODE == "ts_sdk":
compose_file_names.append("ts.yml")
_setup_ts_sdk_integration(dot_env_file, tmp_dir)
diff --git a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
index e389a8bc6b9..cbba1dde33f 100644
--- a/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
+++ b/airflow-e2e-tests/tests/airflow_e2e_tests/constants.py
@@ -53,6 +53,11 @@ KAFKA_DIR_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" /
"docker" / "kafka"
# OpenLineage E2E test paths. The DAGs are sourced from the provider system
tests at runtime by
# openlineage_tests/prepare_dags.py; the overlay carries the
OpenLineage-specific env + dag_doc mount.
OPENLINEAGE_COMPOSE_PATH = AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker"
/ "openlineage.yml"
+# Pins the metadata-DB driver back to psycopg2 for the compat matrix
(--airflow-version), whose
+# released base images predate the psycopg3 default; see
openlineage-compat-db.yml for details.
+OPENLINEAGE_COMPAT_DB_COMPOSE_PATH = (
+ AIRFLOW_ROOT_PATH / "airflow-e2e-tests" / "docker" /
"openlineage-compat-db.yml"
+)
# CI sets this (the same switch the lang-SDK k8s job uses) to build the
lang-SDK
# artifacts with the host toolchain instead of ephemeral toolchain containers.
diff --git a/chart/values.yaml b/chart/values.yaml
index 06330ed41ed..a1c057239d8 100644
--- a/chart/values.yaml
+++ b/chart/values.yaml
@@ -445,12 +445,14 @@ data:
# connection: base64_encoded_connection_string
#
# The 'connection' key is base64-encoded SQLAlchemy connection string, e.g.:
- # postgresql+psycopg2://airflow:password@postgres/airflow
+ # postgresql+psycopg://airflow:password@postgres/airflow (Airflow
>=3.4.0)
+ # postgresql+psycopg2://airflow:password@postgres/airflow (Airflow
<3.4.0, or with the [psycopg2] extra)
metadataSecretName: ~
# If not set, falls back to metadataSecretName. The secret must contain
'connection' key which is
# a base64-encoded connection string, e.g.:
- # postgresql+psycopg2://user:password@host/db
+ # postgresql+psycopg://user:password@host/db (Airflow >=3.4.0)
+ # postgresql+psycopg2://user:password@host/db (Airflow <3.4.0, or with
the [psycopg2] extra)
resultBackendSecretName: ~
brokerUrlSecretName: ~
diff --git a/scripts/ci/docker-compose/backend-postgres.yml
b/scripts/ci/docker-compose/backend-postgres.yml
index d2971bc1e62..b1a64f16a01 100644
--- a/scripts/ci/docker-compose/backend-postgres.yml
+++ b/scripts/ci/docker-compose/backend-postgres.yml
@@ -19,8 +19,8 @@ services:
airflow:
environment:
- BACKEND=postgres
- -
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://postgres:airflow@postgres/airflow
- -
AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql+psycopg2://postgres:airflow@postgres/airflow
+ -
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql://postgres:airflow@postgres/airflow
+ -
AIRFLOW__CELERY__RESULT_BACKEND=db+postgresql+psycopg://postgres:airflow@postgres/airflow
depends_on:
postgres:
condition: service_healthy
diff --git a/task-sdk-integration-tests/docker-compose.yaml
b/task-sdk-integration-tests/docker-compose.yaml
index 82143e98b5e..b71de00d27e 100644
--- a/task-sdk-integration-tests/docker-compose.yaml
+++ b/task-sdk-integration-tests/docker-compose.yaml
@@ -27,7 +27,7 @@ x-airflow-common:
# yamllint disable rule:line-length
AIRFLOW__CORE__AUTH_MANAGER:
'airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager'
AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: 'true'
- AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:
postgresql+psycopg2://airflow:airflow@postgres/airflow
+ AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:
postgresql+psycopg://airflow:airflow@postgres/airflow
AIRFLOW__CORE__FERNET_KEY: ${FERNET_KEY}
AIRFLOW__CORE__LOAD_EXAMPLES: 'false'
AIRFLOW__CORE__DAGS_FOLDER: '/opt/airflow/dags'
