This is an automated email from the ASF dual-hosted git repository.
dabla 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 484a87b54bc Add get_async_extra_dejson to the common.compat connection
helpers (#74147)
484a87b54bc is described below
commit 484a87b54bcea2cbb38700de340e2bc93f4b06fd
Author: David Blain <[email protected]>
AuthorDate: Sat Oct 3 15:35:37 2026 +0200
Add get_async_extra_dejson to the common.compat connection helpers (#74147)
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 <[email protected]>
---
.../providers/common/compat/connection/__init__.py | 30 ++++++++++-
.../common/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 f37c09199d9..a8d1d7bceba 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 079608e765f..cca61658865 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 @@ class TestGetAsyncConnection:
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
