From ab87f492342a7c6a0e5aef57c0b66e9b44bd6245 Mon Sep 17 00:00:00 2001 From: David Blain Date: Sat, 3 Oct 2026 14:44:23 +0200 Subject: [PATCH 1/3] Add get_async_extra_dejson to the common.compat connection helpers Connection.extra_dejson masks the extra's secrets with a synchronous call to the supervisor, which raises DeadlockImminentError when a hook reads it on an event loop with another async call in flight. Airflow 3.3.2+ has Connection.aextra_dejson() (#71890), but providers that still support older versions cannot call it directly, so async hooks fall back to json.loads(conn.extra) and skip the masking (#72130). get_async_extra_dejson(conn) awaits Connection.aextra_dejson() when it exists, and otherwise runs extra_dejson in a worker thread, where blocking on the supervisor is safe, as get_async_connection() does for get_connection(). Co-Authored-By: Claude Opus 5.5 --- .../common/compat/connection/__init__.py | 30 +++++++++- .../compat/connection/test_connection.py | 58 ++++++++++++++++++- 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/providers/common/compat/src/airflow/providers/common/compat/connection/__init__.py b/providers/common/compat/src/airflow/providers/common/compat/connection/__init__.py index f37c09199d9bc..a8d1d7bcebadf 100644 --- a/providers/common/compat/src/airflow/providers/common/compat/connection/__init__.py +++ b/providers/common/compat/src/airflow/providers/common/compat/connection/__init__.py @@ -18,7 +18,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from airflow.providers.common.compat.sdk import BaseHook @@ -50,6 +50,34 @@ async def get_async_connection(conn_id: str, hook: BaseHook | type[BaseHook] | N return await sync_to_async(hook.get_connection)(conn_id=conn_id) +async def get_async_extra_dejson(conn: Connection) -> dict[str, Any]: + """ + Get the connection's extra as a dict, asynchronously and backwards compatible. + + The async counterpart of ``Connection.extra_dejson``, for hooks running on an event loop + (async tasks, triggers). ``extra_dejson`` masks the extra's secrets with a synchronous + call to the supervisor, which raises ``DeadlockImminentError`` when another async call + is in flight on the same event loop. + + On Airflow 3.3.2+ this awaits ``Connection.aextra_dejson()``, which masks the secrets + asynchronously. On older versions the synchronous ``extra_dejson`` runs in a worker + thread, where blocking on the supervisor is safe, the same way + :func:`get_async_connection` falls back to ``get_connection``. + + :param conn: The connection, e.g. from :func:`get_async_connection`. + :returns: The deserialized extra, with its secrets masked. + """ + if hasattr(conn, "aextra_dejson"): + log.debug("Get connection extra using `Connection.aextra_dejson()`.") + return await conn.aextra_dejson() + + from asgiref.sync import sync_to_async + + log.debug("Get connection extra using `Connection.extra_dejson` in a worker thread.") + return await sync_to_async(lambda: conn.extra_dejson)() + + __all__ = [ "get_async_connection", + "get_async_extra_dejson", ] diff --git a/providers/common/compat/tests/unit/common/compat/connection/test_connection.py b/providers/common/compat/tests/unit/common/compat/connection/test_connection.py index 079608e765fef..cca6165886528 100644 --- a/providers/common/compat/tests/unit/common/compat/connection/test_connection.py +++ b/providers/common/compat/tests/unit/common/compat/connection/test_connection.py @@ -16,13 +16,15 @@ # under the License. from __future__ import annotations +import json import logging +import threading from unittest import mock import pytest from airflow.models.connection import Connection -from airflow.providers.common.compat.connection import get_async_connection +from airflow.providers.common.compat.connection import get_async_connection, get_async_extra_dejson class MockAgetBaseHook: @@ -89,3 +91,57 @@ def get_connection(cls, conn_id: str): conn = await get_async_connection("test_conn", hook=OverrideHook) assert conn.password == "override_token" + + +def _raising_extra_dejson(): + return mock.PropertyMock(side_effect=AssertionError("extra_dejson must not run on the event loop")) + + +class TestGetAsyncExtraDejson: + @pytest.mark.asyncio + async def test_uses_aextra_dejson_when_available(self, caplog): + """Airflow 3.3.2+: the extra comes from ``Connection.aextra_dejson()``, masked asynchronously.""" + conn = mock.Mock(spec=["extra", "extra_dejson", "aextra_dejson"]) + conn.aextra_dejson = mock.AsyncMock(return_value={"api_key": "secret"}) + type(conn).extra_dejson = _raising_extra_dejson() + + with caplog.at_level(logging.DEBUG): + extra = await get_async_extra_dejson(conn) + + assert extra == {"api_key": "secret"} + conn.aextra_dejson.assert_awaited_once_with() + assert "Get connection extra using `Connection.aextra_dejson()`." in caplog.text + + @pytest.mark.asyncio + async def test_falls_back_to_extra_dejson_in_a_worker_thread(self, caplog): + """Older Airflow: ``extra_dejson`` runs off the event loop thread, where its sync masking is safe.""" + conn = mock.Mock(spec=["extra", "extra_dejson"]) + loop_thread = threading.get_ident() + threads = [] + + def extra_dejson(): + threads.append(threading.get_ident()) + return {"api_key": "secret"} + + type(conn).extra_dejson = mock.PropertyMock(side_effect=extra_dejson) + + with caplog.at_level(logging.DEBUG): + extra = await get_async_extra_dejson(conn) + + assert extra == {"api_key": "secret"} + assert len(threads) == 1 + assert threads[0] != loop_thread + assert "Get connection extra using `Connection.extra_dejson` in a worker thread." in caplog.text + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("extra", "expected"), + [ + pytest.param(None, {}, id="no-extra"), + pytest.param(json.dumps({"timeout": 30}), {"timeout": 30}, id="extra"), + ], + ) + async def test_with_a_connection(self, extra, expected): + conn = Connection(conn_id="test_conn", conn_type="http", extra=extra) + + assert await get_async_extra_dejson(conn) == expected From 1705f49ddcc200a8edc99280efad2bbad710c368 Mon Sep 17 00:00:00 2001 From: David Blain Date: Sat, 3 Oct 2026 15:10:27 +0200 Subject: [PATCH 2/3] Read the msgraph connection extra with get_async_extra_dejson KiotaRequestAdapterHook.get_async_conn() deserialized the raw extra with json.loads(connection.extra) to avoid extra_dejson, whose secret masking is a synchronous call on the event loop. That skipped masking the extra's values. It now uses common.compat's get_async_extra_dejson(), which masks them asynchronously on Airflow 3.3.2+ and from a worker thread on older versions. The deprecated sync get_conn() is unchanged. Co-Authored-By: Claude Opus 5.5 --- providers/microsoft/azure/pyproject.toml | 2 +- .../providers/microsoft/azure/hooks/msgraph.py | 16 +++++++++------- .../unit/microsoft/azure/hooks/test_msgraph.py | 17 +++++++++++++++++ 3 files changed, 27 insertions(+), 8 deletions(-) diff --git a/providers/microsoft/azure/pyproject.toml b/providers/microsoft/azure/pyproject.toml index d921466ed5d97..6d8ca19e9a6c0 100644 --- a/providers/microsoft/azure/pyproject.toml +++ b/providers/microsoft/azure/pyproject.toml @@ -60,7 +60,7 @@ requires-python = ">=3.10" # After you modify the dependencies, and rebuild your Breeze CI image with ``breeze ci-image build`` dependencies = [ "apache-airflow>=2.11.0", - "apache-airflow-providers-common-compat>=1.13.0", + "apache-airflow-providers-common-compat>=1.13.0", # use next version "adlfs>=2026.5.0", "aiohttp>=3.14.0", "azure-batch>=15.0.0", diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py index 0fc2de4403ce6..fe51c5ab7c7ed 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py +++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py @@ -67,7 +67,7 @@ from msgraph_core._enums import NationalClouds from airflow.exceptions import AirflowBadRequest, AirflowConfigException, AirflowProviderDeprecationWarning -from airflow.providers.common.compat.connection import get_async_connection +from airflow.providers.common.compat.connection import get_async_connection, get_async_extra_dejson from airflow.providers.common.compat.sdk import AirflowException, AirflowNotFoundException, BaseHook, redact if TYPE_CHECKING: @@ -381,12 +381,9 @@ def get_allowed_hosts(authority: str | None, config: dict) -> list[str]: return [] return [host for host in allowed_hosts.split(",") if host] - def _build_request_adapter(self, connection) -> tuple[str, RequestAdapter]: + def _build_request_adapter(self, connection, config: dict[str, Any]) -> tuple[str, RequestAdapter]: client_id = connection.login client_secret = connection.password - # TODO (#54350): do not use connection.extra_dejson until it's fixed in Airflow otherwise expect: - # RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop. - config = json.loads(connection.extra) if connection.extra else {} api_version = self.get_api_version(config) host = self.get_host(connection) # type: ignore[arg-type] base_url = self.get_base_url(host, api_version, config) @@ -473,7 +470,10 @@ def get_conn(self) -> RequestAdapter: if not request_adapter: connection = self.get_connection(conn_id=self.conn_id) - api_version, request_adapter = self._build_request_adapter(connection) + # TODO (#54350): do not use connection.extra_dejson until it's fixed in Airflow otherwise expect: + # RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop. + config = json.loads(connection.extra) if connection.extra else {} + api_version, request_adapter = self._build_request_adapter(connection, config) self.cached_request_adapters[self.conn_id] = (api_version, request_adapter) self.api_version = api_version return request_adapter @@ -509,7 +509,9 @@ async def get_async_conn(self) -> RequestAdapter: if not request_adapter: connection = await get_async_connection(conn_id=self.conn_id) - api_version, request_adapter = self._build_request_adapter(connection) + # Masks the extra's secrets without a synchronous call to the Task SDK on the event loop. + config = await get_async_extra_dejson(connection) + api_version, request_adapter = self._build_request_adapter(connection, config) self.cached_request_adapters[self.conn_id] = (api_version, request_adapter) self.api_version = api_version diff --git a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py index 8ddf8de6dea1c..f3998f18c620d 100644 --- a/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py +++ b/providers/microsoft/azure/tests/unit/microsoft/azure/hooks/test_msgraph.py @@ -531,6 +531,23 @@ async def test_assert_allowed_host_accepts_a_host_listed_in_the_connection(self) await hook.assert_allowed_host("https://other.example/v1.0/users") + @pytest.mark.asyncio + async def test_get_async_conn_reads_the_extra_with_get_async_extra_dejson(self): + """The extra comes from the async compat helper, which masks its secrets off the sync Task SDK path.""" + with ( + patch_hook(), + patch( + "airflow.providers.microsoft.azure.hooks.msgraph.get_async_extra_dejson", + new_callable=AsyncMock, + side_effect=lambda connection: json.loads(connection.extra), + ) as mock_get_async_extra_dejson, + ): + hook = KiotaRequestAdapterHook(conn_id="msgraph_async_extra") + await hook.get_async_conn() + + mock_get_async_extra_dejson.assert_awaited_once() + assert mock_get_async_extra_dejson.await_args.args[0].conn_id == "msgraph_async_extra" + @pytest.mark.asyncio async def test_build_request_adapter_masks_secrets(self): """Test that sensitive data is masked when building request adapter.""" From 92a8f34e190e52b9fa8ccedde719b3daeed01742 Mon Sep 17 00:00:00 2001 From: David Blain Date: Sat, 3 Oct 2026 15:29:51 +0200 Subject: [PATCH 3/3] Use extra_dejson in the msgraph sync get_conn now #54350 is fixed Co-Authored-By: Claude Opus 5.5 --- .../src/airflow/providers/microsoft/azure/hooks/msgraph.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py index fe51c5ab7c7ed..31abb53e1fc6b 100644 --- a/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py +++ b/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py @@ -470,10 +470,7 @@ def get_conn(self) -> RequestAdapter: if not request_adapter: connection = self.get_connection(conn_id=self.conn_id) - # TODO (#54350): do not use connection.extra_dejson until it's fixed in Airflow otherwise expect: - # RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop. - config = json.loads(connection.extra) if connection.extra else {} - api_version, request_adapter = self._build_request_adapter(connection, config) + api_version, request_adapter = self._build_request_adapter(connection, connection.extra_dejson) self.cached_request_adapters[self.conn_id] = (api_version, request_adapter) self.api_version = api_version return request_adapter