Read the msgraph connection extra with get_async_extra_dejson

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",

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,7 @@ 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)
+            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
@@ -509,7 +506,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

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."""