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 12bc0146bb7 Invalidate cached Microsoft Graph request adapter on 401
Unauthorized (#72688)
12bc0146bb7 is described below
commit 12bc0146bb795029830e10944be05fdfdbaca4b3
Author: David Blain <[email protected]>
AuthorDate: Tue Sep 8 17:35:02 2026 +0200
Invalidate cached Microsoft Graph request adapter on 401 Unauthorized
(#72688)
* Invalidate cached Microsoft Graph request adapter on 401 Unauthorized
* Close cached MSGraph request adapter and credential on request failure
---
.../providers/microsoft/azure/hooks/msgraph.py | 22 ++++++-
.../unit/microsoft/azure/hooks/test_msgraph.py | 69 +++++++++++++++++++++-
2 files changed, 88 insertions(+), 3 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 db736169bc3..b81f1aacba5 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
@@ -171,6 +171,8 @@ class DefaultResponseHandler(ResponseHandler):
status_code = HTTPStatus(resp.status_code)
if status_code == HTTPStatus.BAD_REQUEST:
raise AirflowBadRequest(message)
+ if status_code == HTTPStatus.UNAUTHORIZED:
+ raise PermissionError(message)
if status_code == HTTPStatus.NOT_FOUND:
raise AirflowNotFoundException(message)
raise AirflowException(message)
@@ -720,15 +722,31 @@ class KiotaRequestAdapterHook(BaseHook):
request_info=request_info,
error_map=self.error_mapping(),
)
- except (RuntimeError, ValueError) as e:
+ except (PermissionError, RuntimeError, ValueError) as e:
self.log.warning(
"Request failed for conn_id '%s': %s. Invalidating cached
request adapter.",
self.conn_id,
e,
)
- self.cached_request_adapters.pop(self.conn_id, None)
+ await self.close()
raise
+ async def close(self) -> None:
+ """Close the request adapter cached for this connection and evict it
from the cache."""
+ _, request_adapter = self.cached_request_adapters.pop(self.conn_id,
(None, None))
+
+ if not request_adapter:
+ return
+
+ try:
+ adapter = cast("HttpxRequestAdapter", request_adapter)
+ await adapter._http_client.aclose()
+ finally:
+ provider = cast("BaseBearerTokenAuthenticationProvider",
adapter._authentication_provider)
+ access_token_provider = cast("AzureIdentityAccessTokenProvider",
provider.access_token_provider)
+ credential = cast("CachedAsyncTokenCredential",
access_token_provider._credentials)
+ await credential._credential.close()
+
def request_information(
self,
url: str,
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 b505055ace4..0a8dead772b 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
@@ -679,7 +679,7 @@ class TestKiotaRequestAdapterHook:
@pytest.mark.asyncio
async def
test_send_request_invalidates_cache_and_raises_on_any_error(self):
- """send_request evicts the cached adapter and re-raises on any request
error."""
+ """send_request evicts the cached adapter, closes it, and re-raises on
any request error."""
with patch_hook():
hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
@@ -690,11 +690,72 @@ class TestKiotaRequestAdapterHook:
adapter.send_no_response_content_async =
AsyncMock(side_effect=RuntimeError("some error"))
hook.cached_request_adapters[hook.conn_id] = (hook.api_version,
adapter)
+ access_token_provider =
adapter._authentication_provider.access_token_provider
+ credential = access_token_provider._credentials._credential
+
with pytest.raises(RuntimeError, match="some error"):
await hook.run(url="users")
adapter.send_no_response_content_async.assert_called_once()
assert hook.conn_id not in hook.cached_request_adapters
+ adapter._http_client.aclose.assert_awaited_once()
+ credential.close.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def
test_send_request_invalidates_cache_and_raises_on_unauthorized(self):
+ """send_request evicts the cached adapter, closes it, and re-raises
when Microsoft Graph returns 401."""
+ with patch_hook():
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ adapter = Mock(spec=HttpxRequestAdapter)
+ adapter._http_client = Mock(spec=AsyncClient, is_closed=False)
+ adapter._authentication_provider =
mock_authentication_provider(closed=False)
+ adapter.base_url = "https://graph.microsoft.com/v1.0";
+ adapter.send_no_response_content_async = AsyncMock(
+ side_effect=PermissionError("401 Unauthorized")
+ )
+ hook.cached_request_adapters[hook.conn_id] = (hook.api_version,
adapter)
+
+ access_token_provider =
adapter._authentication_provider.access_token_provider
+ credential = access_token_provider._credentials._credential
+
+ with pytest.raises(PermissionError, match="401 Unauthorized"):
+ await hook.run(url="users")
+
+ adapter.send_no_response_content_async.assert_called_once()
+ assert hook.conn_id not in hook.cached_request_adapters
+ adapter._http_client.aclose.assert_awaited_once()
+ credential.close.assert_awaited_once()
+
+ @pytest.mark.asyncio
+ async def test_close_closes_http_client_and_credential(self):
+ """close() closes the cached HTTP client and the underlying
credential, then evicts the cache."""
+ with patch_hook():
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ adapter = Mock(spec=HttpxRequestAdapter)
+ adapter._http_client = Mock(spec=AsyncClient, is_closed=False)
+ adapter._authentication_provider =
mock_authentication_provider(closed=False)
+ hook.cached_request_adapters[hook.conn_id] = (hook.api_version,
adapter)
+
+ access_token_provider =
adapter._authentication_provider.access_token_provider
+ credential = access_token_provider._credentials._credential
+
+ await hook.close()
+
+ adapter._http_client.aclose.assert_awaited_once()
+ credential.close.assert_awaited_once()
+ assert hook.conn_id not in hook.cached_request_adapters
+
+ @pytest.mark.asyncio
+ async def test_close_is_a_no_op_when_nothing_is_cached(self):
+ """close() does nothing when there is no cached request adapter for
the conn_id."""
+ with patch_hook():
+ hook = KiotaRequestAdapterHook(conn_id="msgraph_api")
+
+ await hook.close()
+
+ assert hook.conn_id not in hook.cached_request_adapters
def test_allowed_hosts_is_empty_list_when_not_configured(self):
"""An unset allowed_hosts/authority must yield []."""
@@ -842,6 +903,12 @@ class TestResponseHandler:
with pytest.raises(AirflowBadRequest):
asyncio.run(DefaultResponseHandler().handle_response_async(response, None))
+ def test_handle_response_async_when_unauthorized(self):
+ response = mock_json_response(401, {})
+
+ with pytest.raises(PermissionError):
+
asyncio.run(DefaultResponseHandler().handle_response_async(response, None))
+
def test_handle_response_async_when_not_found(self):
response = mock_json_response(404, {})
