From 8723774eb69672282bdecff4796e91a6d768dd48 Mon Sep 17 00:00:00 2001 From: Dev-iL <6509619+Dev-iL@users.noreply.github.com> Date: Mon, 6 Jul 2026 10:56:28 +0300 Subject: [PATCH 1/3] Fix celery result-backend URL derivation for the new Postgres default The hardcoded fallback that derives [celery] result_backend from sql_alchemy_conn when a driverless postgresql:// URL is configured targeted postgresql+psycopg2:// explicitly. Update it to postgresql+psycopg:// to match the sync Postgres driver default change in apache-airflow and apache-airflow-providers-postgres. Part of the migration tracked in apache/airflow#68453. --- providers/celery/docs/changelog.rst | 7 +++++++ providers/celery/provider.yaml | 2 +- .../providers/celery/executors/default_celery.py | 6 +++--- .../providers/celery/get_provider_info.py | 2 +- .../celery/executors/test_celery_executor.py | 16 ++++++++++++++++ 5 files changed, 28 insertions(+), 5 deletions(-) diff --git a/providers/celery/docs/changelog.rst b/providers/celery/docs/changelog.rst index b0650015fa0ab..5d1e48876e85f 100644 --- a/providers/celery/docs/changelog.rst +++ b/providers/celery/docs/changelog.rst @@ -27,6 +27,13 @@ Changelog --------- +.. note:: + When ``[celery] result_backend`` is not set and Airflow derives it from ``sql_alchemy_conn``, a + driverless ``postgresql://`` connection string now produces ``db+postgresql+psycopg://...`` instead + of ``db+postgresql+psycopg2://...``, matching the sync Postgres driver default change in + ``apache-airflow`` and ``apache-airflow-providers-postgres``. An explicit ``[celery] result_backend`` + or an explicit driver in ``sql_alchemy_conn`` is unaffected. + 3.21.0 ...... diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml index e10f5030bd470..a5e996df4b597 100644 --- a/providers/celery/provider.yaml +++ b/providers/celery/provider.yaml @@ -247,7 +247,7 @@ config: version_added: ~ type: string sensitive: true - example: "db+postgresql+psycopg2://postgres:airflow@postgres/airflow" + example: "db+postgresql+psycopg://postgres:airflow@postgres/airflow" default: ~ result_backend_sqlalchemy_engine_options: description: | diff --git a/providers/celery/src/airflow/providers/celery/executors/default_celery.py b/providers/celery/src/airflow/providers/celery/executors/default_celery.py index 9dd94036734e0..678f12293bb2e 100644 --- a/providers/celery/src/airflow/providers/celery/executors/default_celery.py +++ b/providers/celery/src/airflow/providers/celery/executors/default_celery.py @@ -79,9 +79,9 @@ def get_default_celery_config(team_conf) -> dict[str, Any]: else: log.debug("Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.") sql_alchemy_conn = team_conf.get("database", "SQL_ALCHEMY_CONN") - # In SQLAlchemy 2.1 the default PostgreSQL driver changed from psycopg2 to psycopg (v3). - # To maintain existing behavior, we explicitly specify psycopg2 for driverless PostgreSQL URLs. - sql_alchemy_conn = sql_alchemy_conn.replace("postgresql://", "postgresql+psycopg2://", 1) + # Airflow's sync metadata-database driver default is psycopg (v3); mirror that explicitly + # for driverless PostgreSQL URLs instead of relying on SQLAlchemy's own default. + sql_alchemy_conn = sql_alchemy_conn.replace("postgresql://", "postgresql+psycopg://", 1) result_backend = f"db+{sql_alchemy_conn}" # Handle result backend transport options (for Redis Sentinel support) diff --git a/providers/celery/src/airflow/providers/celery/get_provider_info.py b/providers/celery/src/airflow/providers/celery/get_provider_info.py index a715038d84143..4a79cc58821d5 100644 --- a/providers/celery/src/airflow/providers/celery/get_provider_info.py +++ b/providers/celery/src/airflow/providers/celery/get_provider_info.py @@ -130,7 +130,7 @@ def get_provider_info(): "version_added": None, "type": "string", "sensitive": True, - "example": "db+postgresql+psycopg2://postgres:airflow@postgres/airflow", + "example": "db+postgresql+psycopg://postgres:airflow@postgres/airflow", "default": None, }, "result_backend_sqlalchemy_engine_options": { diff --git a/providers/celery/tests/unit/celery/executors/test_celery_executor.py b/providers/celery/tests/unit/celery/executors/test_celery_executor.py index c349efa32cb4e..f61aabc8fd82f 100644 --- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py +++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py @@ -789,6 +789,22 @@ def test_result_backend_transport_options_with_multiple_options(): assert result_backend_opts["master_name"] == "mymaster" +@conf_vars( + { + ("celery", "result_backend"): None, + ("database", "sql_alchemy_conn"): "postgresql://user:pass@host/db", + } +) +def test_result_backend_derived_from_sql_alchemy_conn_uses_psycopg(): + """A driverless sql_alchemy_conn must derive a psycopg (v3) result_backend, not psycopg2.""" + import importlib + + importlib.reload(default_celery) + assert ( + default_celery.DEFAULT_CELERY_CONFIG["result_backend"] == "db+postgresql+psycopg://user:pass@host/db" + ) + + @conf_vars({("celery_result_backend_transport_options", "sentinel_kwargs"): "invalid_json"}) def test_result_backend_sentinel_kwargs_invalid_json(): """Test that invalid JSON in sentinel_kwargs raises an error.""" From 7ac638a2ad31ce4b329d7b62d36f21624b0df62e Mon Sep 17 00:00:00 2001 From: Dev-iL <6509619+Dev-iL@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:45:46 +0300 Subject: [PATCH 2/3] Guard celery's psycopg default derivation against older environments The result-backend URL derivation unconditionally targeted postgresql+psycopg://, but that dialect requires SQLAlchemy 2.0+ and psycopg installed. providers/celery's own dependency floor is unbumped, so an environment still on the older combination would have its result-backend derivation silently start failing on upgrade. Gate the target scheme on the same psycopg3-availability check PostgresHook itself uses, falling back to psycopg2 otherwise. --- .../celery/executors/default_celery.py | 24 +++++++++++++++++-- .../celery/executors/test_celery_executor.py | 13 ++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/providers/celery/src/airflow/providers/celery/executors/default_celery.py b/providers/celery/src/airflow/providers/celery/executors/default_celery.py index 678f12293bb2e..3d90b827d5123 100644 --- a/providers/celery/src/airflow/providers/celery/executors/default_celery.py +++ b/providers/celery/src/airflow/providers/celery/executors/default_celery.py @@ -31,6 +31,23 @@ log = logging.getLogger(__name__) +_USE_PSYCOPG3: bool +try: + from importlib.util import find_spec + + import sqlalchemy + from packaging.version import Version + + _sqlalchemy_version = Version(sqlalchemy.__version__) + _is_sqla2 = (_sqlalchemy_version.major, _sqlalchemy_version.minor, _sqlalchemy_version.micro) >= ( + 2, + 0, + 0, + ) + _USE_PSYCOPG3 = find_spec("psycopg") is not None and _is_sqla2 +except (ImportError, ModuleNotFoundError): + _USE_PSYCOPG3 = False + def _broker_supports_visibility_timeout(url): return url.startswith(("redis://", "rediss://", "sqs://", "sentinel://")) @@ -80,8 +97,11 @@ def get_default_celery_config(team_conf) -> dict[str, Any]: log.debug("Value for celery result_backend not found. Using sql_alchemy_conn with db+ prefix.") sql_alchemy_conn = team_conf.get("database", "SQL_ALCHEMY_CONN") # Airflow's sync metadata-database driver default is psycopg (v3); mirror that explicitly - # for driverless PostgreSQL URLs instead of relying on SQLAlchemy's own default. - sql_alchemy_conn = sql_alchemy_conn.replace("postgresql://", "postgresql+psycopg://", 1) + # for driverless PostgreSQL URLs instead of relying on SQLAlchemy's own default. Fall back to + # psycopg2 when psycopg/SQLAlchemy 2.0 aren't both available, matching PostgresHook's own + # USE_PSYCOPG3 gating so this doesn't break environments still on the older combination. + target_scheme = "postgresql+psycopg://" if _USE_PSYCOPG3 else "postgresql+psycopg2://" + sql_alchemy_conn = sql_alchemy_conn.replace("postgresql://", target_scheme, 1) result_backend = f"db+{sql_alchemy_conn}" # Handle result backend transport options (for Redis Sentinel support) diff --git a/providers/celery/tests/unit/celery/executors/test_celery_executor.py b/providers/celery/tests/unit/celery/executors/test_celery_executor.py index f61aabc8fd82f..4013479efdec8 100644 --- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py +++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py @@ -805,6 +805,19 @@ def test_result_backend_derived_from_sql_alchemy_conn_uses_psycopg(): ) +@conf_vars( + { + ("celery", "result_backend"): None, + ("database", "sql_alchemy_conn"): "postgresql://user:pass@host/db", + } +) +def test_result_backend_falls_back_to_psycopg2_without_psycopg3(monkeypatch): + """Without psycopg/SQLAlchemy 2.0 available, the derivation must fall back to psycopg2.""" + monkeypatch.setattr(default_celery, "_USE_PSYCOPG3", False) + config = default_celery.get_default_celery_config(conf) + assert config["result_backend"] == "db+postgresql+psycopg2://user:pass@host/db" + + @conf_vars({("celery_result_backend_transport_options", "sentinel_kwargs"): "invalid_json"}) def test_result_backend_sentinel_kwargs_invalid_json(): """Test that invalid JSON in sentinel_kwargs raises an error.""" From 78e84d4c16e1b87487f998cb6782d95d1fe9c25e Mon Sep 17 00:00:00 2001 From: Dev-iL <6509619+Dev-iL@users.noreply.github.com> Date: Mon, 6 Jul 2026 12:09:57 +0300 Subject: [PATCH 3/3] Fix non-deterministic celery result-backend derivation test test_result_backend_derived_from_sql_alchemy_conn_uses_psycopg asserted the psycopg3 scheme unconditionally, but the derivation now depends on _USE_PSYCOPG3, which reflects whatever happens to be installed. In a lowest-dependencies environment (providers/celery has no dependency on psycopg or SQLAlchemy 2.0), this test would deterministically fail since the code correctly falls back to psycopg2 there. Pin _USE_PSYCOPG3 explicitly and call get_default_celery_config directly, matching the fallback test added alongside it. --- .../unit/celery/executors/test_celery_executor.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/providers/celery/tests/unit/celery/executors/test_celery_executor.py b/providers/celery/tests/unit/celery/executors/test_celery_executor.py index 4013479efdec8..a4d11a0de0a82 100644 --- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py +++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py @@ -795,14 +795,11 @@ def test_result_backend_transport_options_with_multiple_options(): ("database", "sql_alchemy_conn"): "postgresql://user:pass@host/db", } ) -def test_result_backend_derived_from_sql_alchemy_conn_uses_psycopg(): +def test_result_backend_derived_from_sql_alchemy_conn_uses_psycopg(monkeypatch): """A driverless sql_alchemy_conn must derive a psycopg (v3) result_backend, not psycopg2.""" - import importlib - - importlib.reload(default_celery) - assert ( - default_celery.DEFAULT_CELERY_CONFIG["result_backend"] == "db+postgresql+psycopg://user:pass@host/db" - ) + monkeypatch.setattr(default_celery, "_USE_PSYCOPG3", True) + config = default_celery.get_default_celery_config(conf) + assert config["result_backend"] == "db+postgresql+psycopg://user:pass@host/db" @conf_vars(