Summary

Adds BaseHook.aget_hook() — the missing async counterpart of BaseHook.get_hook() — and amask_secret() — the async counterpart of mask_secret() — for use in async tasks so we don't block the event loop.

Root cause

Async tasks that called BaseHook.get_hook() from inside async def / aexecute() were
inadvertently using the sync connection-fetch path:

# Inside an async @task — wrong
hook = KiotaRequestAdapterHook.get_hook(conn_id=MSGRAPH_CONN_ID)

get_hook() calls get_connection() → Connection.get() → _get_connection(), which
ultimately calls comms.send() — the blocking supervisor round-trip — on the event loop
thread. Blocking the event loop thread while other coroutines are concurrently in-flight via
comms.asend() causes a deadlock or severe stall.

Even after switching to aget_hook(), a second deadlock source remained: _async_get_connection()
called the sync mask_secret() which internally uses comms.send(), triggering DeadlockImminentError.

Changes

1. BaseHook.aget_hook() (task-sdk/src/airflow/sdk/bases/hook.py)

Provides a proper async alternative so user code never needs to call the blocking sync path
from inside a coroutine:

# Before — sync, blocks event loop
hook = KiotaRequestAdapterHook.get_hook(conn_id=MSGRAPH_CONN_ID)

# After — async, uses aget_connection() → asend() → no blocking
hook = await KiotaRequestAdapterHook.aget_hook(conn_id=MSGRAPH_CONN_ID)

aget_hook() delegates to the existing aget_connection(), which already routes through
comms.asend() correctly.

2. amask_secret() (task-sdk/src/airflow/sdk/log.py)

Adds async version of mask_secret() that uses comms.asend() instead of comms.send():

# Sync version — uses comms.send(), unsafe from event loop thread
mask_secret(conn.password)

# Async version — uses comms.asend(), safe in async context
await amask_secret(conn.password)

3. _amask_connection_secrets() / _async_get_connection() (task-sdk/src/airflow/sdk/execution_time/context.py)

• Added _amask_connection_secrets() — async helper that calls amask_secret()
• Updated _async_get_connection() to use await _amask_connection_secrets(conn) instead of the sync _mask_connection_secrets(conn)

This ensures the entire async connection-fetch path is fully non-blocking.

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

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

Claude 4.6

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