Fix DeadlockImminentError when a connection is resolved inside an async task

[dabla]

Fixes a DeadlockImminentError raised in Airflow 3.3.1 when a connection is resolved inside an async task via await BaseHook.aget_hook(...) or await BaseHook.aget_connection(...).

 @task
 async def get_relations(updated_id, **context):
     hook = await KiotaRequestAdapterHook.aget_hook(conn_id=MSGRAPH_CONN_ID)
     return await hook.paginated_run(
           url=f"groups/{updated_id[0]}/owners",
           query_parameters={"$select": "id"},
     )

Problem

The call chain when fetching a connection inside an async task is:

await Connection.async_get()      ← async, running on the event-loop thread
  → Connection.get_uri()          ← sync, still on event-loop thread
      → self.extra_dejson         ← sync property, still on event-loop thread
          → mask_secret()
              → comms.send()      ← ERROR: DeadlockImminentError

The error message says "use the async equivalents" — but the caller already is.
The bug is inside Connection itself: get_uri() and extra_dejson have no async counterparts, so secret masking always goes through the blocking comms.send(). amask_secret() (which uses asend()) already existed in airflow.sdk.log
but nothing in Connection called it.

Changes

task-sdk/src/airflow/sdk/definitions/connection.py

• Extract all URI-assembly logic into a new private _build_uri(self, extra_dejson: dict) helper so sync and async paths share a single implementation.
• Refactor get_uri() to return self._build_uri(self.extra_dejson) — behaviour is identical to before.
• Add aextra_dejson() — async method that mirrors extra_dejson but awaits amask_secret() instead of calling the blocking mask_secret().
• Add aget_uri() — async method that calls await self.aextra_dejson() and delegates to _build_uri().

task-sdk/tests/task_sdk/definitions/test_connection.py

• test_aget_uri — aget_uri() produces the same URI as get_uri() and delegates to aextra_dejson.
• test_aextra_dejson_calls_amask_secret — aextra_dejson() uses amask_secret (async) and never calls the sync mask_secret.
• test_aextra_dejson_no_extra — aextra_dejson() returns {} without calling amask_secret when extra is None.

Usage

# Before — triggers DeadlockImminentError in Airflow 3.3.1:
conn = await Connection.async_get("my_conn")
uri = conn.get_uri()

# After — safe inside any async task:
conn = await Connection.async_get("my_conn")
uri = await conn.aget_uri()

---

Was generative AI tooling used to co-author this PR?

• [ x ] Yes (please specify the tool below)

Claude Sonnet 4.6 (GitHub Copilot)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[amoghrajesh]

Directionally fine, I have a few qns / todos here.

1. Can we also update the doc: https://airflow.apache.org/docs/task-sdk/stable/deferred-vs-async-operators.html? The example mentions hook = KiotaRequestAdapterHook.get_hook(conn_id="msgraph_default")
2. Should we be also covering the other methods like to_dict, get_extra_dejson here?

[dabla]

Directionally fine, I have a few qns / todos here.

1. Can we also update the doc: https://airflow.apache.org/docs/task-sdk/stable/deferred-vs-async-operators.html? The example mentions hook = KiotaRequestAdapterHook.get_hook(conn_id="msgraph_default")
2. Should we be also covering the other methods like to_dict, get_extra_dejson here?

Don't know if we still want to implement an async version of get_extra_dejson knowing it's already deprecated in favor of extra_dejson which already has an async counterpart? I don't think to_dict is used in async paths but we could add it if you want?