Async review pass: fix http/retry bugs, remove asyncio.gather, raise coverage to 98%

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Abel Milash

src/PowerPlatform/Dataverse/aio/__init__.py

@@ -1,4 +1,12 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
+"""
+Async namespace for the PowerPlatform Dataverse SDK.
+
+Import the async client via::
+
+    from PowerPlatform.Dataverse.aio.async_client import AsyncDataverseClient
+"""
+
 __all__ = []

src/PowerPlatform/Dataverse/aio/core/__init__.py

@@ -1,2 +1,9 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
+
+"""
+Async core infrastructure components for the Dataverse SDK.
+
+This module contains the foundational async components including authentication,
+HTTP client, and error handling.
+"""

src/PowerPlatform/Dataverse/aio/core/_async_http.py

@@ -112,16 +112,19 @@ async def _request(self, method: str, url: str, **kwargs: Any) -> aiohttp.Client
         for attempt in range(self.max_attempts):
             try:
                 t0 = time.monotonic()
-                resp = await self._session.request(method, url, **kwargs)
-                # Buffer the full body so the caller can read it freely after this call
-                # returns, and so the connection is released back to the pool.
-                await resp.read()
+                async with self._session.request(method, url, **kwargs) as resp:
+                    await resp.read()
                 elapsed_ms = (time.monotonic() - t0) * 1000
 
                 if self._logger is not None:
                     # Only decode resp.text when body logging is enabled — avoids
                     # unnecessary overhead for large payloads when max_body_bytes == 0.
-                    resp_body = await resp.text() if self._logger.body_logging_enabled else None
+                    resp_body = None
+                    if self._logger.body_logging_enabled:
+                        try:
+                            resp_body = await resp.text()
+                        except Exception:
+                            pass
                     self._logger.log_response(
                         method,
                         url,
@@ -131,7 +134,7 @@ async def _request(self, method: str, url: str, **kwargs: Any) -> aiohttp.Client
                         elapsed_ms=elapsed_ms,
                     )
                 return resp
-            except aiohttp.ClientError as exc:
+            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                 if self._logger is not None:
                     self._logger.log_error(
                         method,

src/PowerPlatform/Dataverse/aio/data/__init__.py

@@ -1,2 +1,9 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
+
+"""
+Async data access layer for the Dataverse SDK.
+
+This module contains async OData protocol handling, CRUD operations, metadata management,
+SQL query functionality, and file upload capabilities.
+"""

src/PowerPlatform/Dataverse/aio/data/_async_batch.py

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import uuid
 from typing import TYPE_CHECKING, Any, Dict, List, Union
 
@@ -251,7 +250,7 @@ async def _resolve_record_delete(self, op: _RecordDelete) -> List[_RawRequest]:
             return []
         if op.use_bulk_delete:
             return [await self._od._build_delete_multiple(op.table, ids)]
-        return list(await asyncio.gather(*[self._od._build_delete(op.table, rid) for rid in ids]))
+        return [await self._od._build_delete(op.table, rid) for rid in ids]
 
     async def _resolve_record_get(self, op: _RecordGet) -> List[_RawRequest]:
         return [
@@ -315,10 +314,10 @@ async def _resolve_table_add_columns(self, op: _TableAddColumns) -> List[_RawReq
     async def _resolve_table_remove_columns(self, op: _TableRemoveColumns) -> List[_RawRequest]:
         columns = [op.columns] if isinstance(op.columns, str) else list(op.columns)
         metadata_id = await self._require_entity_metadata(op.table)
-        attr_metas = await asyncio.gather(*[
-            self._od._get_attribute_metadata(metadata_id, col_name, extra_select="@odata.type,AttributeType")
+        attr_metas = [
+            await self._od._get_attribute_metadata(metadata_id, col_name, extra_select="@odata.type,AttributeType")
             for col_name in columns
-        ])
+        ]
         requests: List[_RawRequest] = []
         for col_name, attr_meta in zip(columns, attr_metas):
             if not attr_meta or not attr_meta.get("MetadataId"):

src/PowerPlatform/Dataverse/aio/data/_async_odata.py

@@ -141,7 +141,7 @@ def _merge(h: Optional[Dict[str, str]]) -> Dict[str, str]:
         if r.status in request_context.expected:
             return r
 
-        response_headers = dict(r.headers) if r.headers else {}
+        response_headers = getattr(r, "headers", {}) or {}
         raw_text = ""
         try:
             raw_text = await r.text()
@@ -368,9 +368,7 @@ async def _upsert_multiple(
             )
         logical_name = table_schema_name.lower()
         lowered_records = [self._lowercase_keys(r) for r in records]
-        converted = await asyncio.gather(*[
-            self._convert_labels_to_ints(table_schema_name, r) for r in lowered_records
-        ])
+        converted = [await self._convert_labels_to_ints(table_schema_name, r) for r in lowered_records]
         targets: List[Dict[str, Any]] = []
         for alt_key, record_processed in zip(alternate_keys, converted):
             alt_key_lower = self._lowercase_keys(alt_key)
@@ -1477,10 +1475,10 @@ async def _delete_columns(
         deleted: List[str] = []
         needs_picklist_flush = False
 
-        attr_metas = await asyncio.gather(*[
-            self._get_attribute_metadata(metadata_id, col, extra_select="@odata.type,AttributeType")
+        attr_metas = [
+            await self._get_attribute_metadata(metadata_id, col, extra_select="@odata.type,AttributeType")
             for col in names
-        ])
+        ]
         for column_name, attr_meta in zip(names, attr_metas):
             if not attr_meta:
                 raise MetadataError(
@@ -1538,7 +1536,7 @@ async def _build_create_multiple(
             raise TypeError("All items for multi-create must be dicts")
         logical_name = table.lower()
         lowered = [self._lowercase_keys(r) for r in records]
-        converted = await asyncio.gather(*[self._convert_labels_to_ints(table, r) for r in lowered])
+        converted = [await self._convert_labels_to_ints(table, r) for r in lowered]
         enriched = [
             {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"} if "@odata.type" not in r else r
             for r in converted
@@ -1591,7 +1589,7 @@ async def _build_update_multiple_from_records(
         """
         logical_name = table.lower()
         lowered = [self._lowercase_keys(r) for r in records]
-        converted = await asyncio.gather(*[self._convert_labels_to_ints(table, r) for r in lowered])
+        converted = [await self._convert_labels_to_ints(table, r) for r in lowered]
         enriched = [
             {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"} if "@odata.type" not in r else r
             for r in converted
@@ -1661,7 +1659,7 @@ async def _build_upsert_multiple(
             )
         logical_name = table.lower()
         lowered_records = [self._lowercase_keys(r) for r in records]
-        converted = await asyncio.gather(*[self._convert_labels_to_ints(table, r) for r in lowered_records])
+        converted = [await self._convert_labels_to_ints(table, r) for r in lowered_records]
         targets: List[Dict[str, Any]] = []
         for alt_key, record_processed in zip(alternate_keys, converted):
             alt_key_lower = self._lowercase_keys(alt_key)

src/PowerPlatform/Dataverse/aio/models/__init__.py

@@ -1,4 +1,13 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
 
+"""
+Async data models and type definitions for the Dataverse SDK.
+
+Provides async-specific models for Dataverse entities:
+
+- :class:`~PowerPlatform.Dataverse.aio.models.async_query_builder.AsyncQueryBuilder`: Async fluent query builder.
+- :class:`~PowerPlatform.Dataverse.aio.models.async_fetchxml_query.AsyncFetchXmlQuery`: Async FetchXML query.
+"""
+
 __all__ = []

tests/unit/aio/core/test_async_http.py

@@ -9,15 +9,34 @@
 from PowerPlatform.Dataverse.aio.core._async_http import _AsyncHttpClient
 
 
-def _make_session(status: int = 200) -> MagicMock:
-    """Return a mock aiohttp.ClientSession whose request() returns a buffered response."""
-    session = MagicMock(spec=aiohttp.ClientSession)
-    resp = AsyncMock()
+def _make_resp(status: int = 200) -> MagicMock:
+    """Return a mock aiohttp.ClientResponse."""
+    resp = MagicMock()
     resp.status = status
     resp.headers = {}
     resp.read = AsyncMock(return_value=b"")
     resp.text = AsyncMock(return_value="")
-    session.request = AsyncMock(return_value=resp)
+    return resp
+
+
+def _make_cm(resp=None, exc=None) -> MagicMock:
+    """Return an async context manager mock.
+
+    If exc is given, __aenter__ raises it. Otherwise it returns resp.
+    """
+    cm = MagicMock()
+    if exc is not None:
+        cm.__aenter__ = AsyncMock(side_effect=exc)
+    else:
+        cm.__aenter__ = AsyncMock(return_value=resp)
+    cm.__aexit__ = AsyncMock(return_value=False)
+    return cm
+
+
+def _make_session(status: int = 200) -> MagicMock:
+    """Return a mock aiohttp.ClientSession whose request() is an async context manager."""
+    session = MagicMock(spec=aiohttp.ClientSession)
+    session.request = MagicMock(return_value=_make_cm(_make_resp(status)))
     return session
 
 
@@ -99,13 +118,11 @@ class TestAsyncHttpClientRetry:
     async def test_retries_on_client_error_and_succeeds(self):
         """Retries after a ClientError and returns response on second attempt."""
         session = MagicMock(spec=aiohttp.ClientSession)
-        good_resp = AsyncMock()
-        good_resp.status = 200
-        good_resp.headers = {}
-        good_resp.read = AsyncMock(return_value=b"")
-        good_resp.text = AsyncMock(return_value="")
-
-        session.request = AsyncMock(side_effect=[aiohttp.ClientConnectionError("timeout"), good_resp])
+        good_resp = _make_resp(200)
+        session.request = MagicMock(side_effect=[
+            _make_cm(exc=aiohttp.ClientConnectionError("timeout")),
+            _make_cm(good_resp),
+        ])
         client = _AsyncHttpClient(retries=2, backoff=0, session=session)
         with patch("asyncio.sleep", new_callable=AsyncMock):
             result = await client._request("get", "https://example.com/data")
@@ -116,7 +133,9 @@ async def test_retries_on_client_error_and_succeeds(self):
     async def test_raises_after_all_retries_exhausted(self):
         """Raises ClientError after all retry attempts fail."""
         session = MagicMock(spec=aiohttp.ClientSession)
-        session.request = AsyncMock(side_effect=aiohttp.ClientConnectionError("timeout"))
+        session.request = MagicMock(
+            return_value=_make_cm(exc=aiohttp.ClientConnectionError("timeout"))
+        )
         client = _AsyncHttpClient(retries=3, backoff=0, session=session)
         with patch("asyncio.sleep", new_callable=AsyncMock):
             with pytest.raises(aiohttp.ClientError):
@@ -125,19 +144,12 @@ async def test_raises_after_all_retries_exhausted(self):
     async def test_backoff_delay_between_retries(self):
         """Sleeps with exponential backoff between retry attempts."""
         session = MagicMock(spec=aiohttp.ClientSession)
-        good_resp = AsyncMock()
-        good_resp.status = 200
-        good_resp.headers = {}
-        good_resp.read = AsyncMock(return_value=b"")
-        good_resp.text = AsyncMock(return_value="")
-
-        session.request = AsyncMock(
-            side_effect=[
-                aiohttp.ClientConnectionError(),
-                aiohttp.ClientConnectionError(),
-                good_resp,
-            ]
-        )
+        good_resp = _make_resp(200)
+        session.request = MagicMock(side_effect=[
+            _make_cm(exc=aiohttp.ClientConnectionError()),
+            _make_cm(exc=aiohttp.ClientConnectionError()),
+            _make_cm(good_resp),
+        ])
         client = _AsyncHttpClient(retries=3, backoff=1.0, session=session)
         with patch("asyncio.sleep", new_callable=AsyncMock) as mock_sleep:
             await client._request("get", "https://example.com/data")
@@ -151,6 +163,22 @@ async def test_no_retry_on_success(self):
         await client._request("get", "https://example.com/data")
         assert session.request.call_count == 1
 
+    async def test_retries_on_timeout_error(self):
+        """Retries on asyncio.TimeoutError (not a subclass of aiohttp.ClientError)."""
+        import asyncio
+        session = MagicMock(spec=aiohttp.ClientSession)
+        good_resp = _make_resp(200)
+        session.request = MagicMock(side_effect=[
+            _make_cm(exc=asyncio.TimeoutError()),
+            _make_cm(good_resp),
+        ])
+        client = _AsyncHttpClient(retries=2, backoff=0, session=session)
+        with patch("asyncio.sleep", new_callable=AsyncMock):
+            result = await client._request("get", "https://example.com/data")
+
+        assert session.request.call_count == 2
+        assert result is good_resp
+
 
 class TestAsyncHttpClientClose:
     """Tests for _AsyncHttpClient.close()."""
@@ -194,15 +222,58 @@ async def test_response_logged_when_logger_set(self):
     async def test_error_logged_on_retry(self):
         """Transport errors are logged before each retry."""
         session = MagicMock(spec=aiohttp.ClientSession)
-        good_resp = AsyncMock()
-        good_resp.status = 200
-        good_resp.headers = {}
-        good_resp.read = AsyncMock(return_value=b"")
-        good_resp.text = AsyncMock(return_value="")
-        session.request = AsyncMock(side_effect=[aiohttp.ClientConnectionError(), good_resp])
+        good_resp = _make_resp(200)
+        session.request = MagicMock(side_effect=[
+            _make_cm(exc=aiohttp.ClientConnectionError()),
+            _make_cm(good_resp),
+        ])
         mock_logger = MagicMock()
         mock_logger.body_logging_enabled = False
         client = _AsyncHttpClient(retries=2, backoff=0, session=session, logger=mock_logger)
         with patch("asyncio.sleep", new_callable=AsyncMock):
             await client._request("get", "https://example.com/data")
         mock_logger.log_error.assert_called_once()
+
+    async def test_request_body_logged_from_json_kwarg(self):
+        """json= kwarg body is extracted and passed to log_request."""
+        session = _make_session()
+        mock_logger = MagicMock()
+        mock_logger.body_logging_enabled = False
+        client = _AsyncHttpClient(retries=1, session=session, logger=mock_logger)
+        await client._request("post", "https://example.com/data", json={"key": "value"})
+        _, log_kwargs = mock_logger.log_request.call_args
+        assert log_kwargs["body"] == {"key": "value"}
+
+    async def test_request_body_logged_from_data_kwarg(self):
+        """data= kwarg body is extracted when json= is absent."""
+        session = _make_session()
+        mock_logger = MagicMock()
+        mock_logger.body_logging_enabled = False
+        client = _AsyncHttpClient(retries=1, session=session, logger=mock_logger)
+        await client._request("post", "https://example.com/data", data=b"raw bytes")
+        _, log_kwargs = mock_logger.log_request.call_args
+        assert log_kwargs["body"] == b"raw bytes"
+
+    async def test_response_body_decoded_when_body_logging_enabled(self):
+        """When body_logging_enabled=True, resp.text() is awaited and passed to log_response."""
+        session = _make_session()
+        session.request.return_value.__aenter__.return_value.text = AsyncMock(return_value='{"ok": true}')
+        mock_logger = MagicMock()
+        mock_logger.body_logging_enabled = True
+        client = _AsyncHttpClient(retries=1, session=session, logger=mock_logger)
+        await client._request("get", "https://example.com/data")
+        _, log_kwargs = mock_logger.log_response.call_args
+        assert log_kwargs["body"] == '{"ok": true}'
+
+    async def test_response_body_decode_error_is_swallowed(self):
+        """If resp.text() raises, body is None and log_response is still called."""
+        session = _make_session()
+        session.request.return_value.__aenter__.return_value.text = AsyncMock(
+            side_effect=UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte")
+        )
+        mock_logger = MagicMock()
+        mock_logger.body_logging_enabled = True
+        client = _AsyncHttpClient(retries=1, session=session, logger=mock_logger)
+        await client._request("get", "https://example.com/data")
+        _, log_kwargs = mock_logger.log_response.call_args
+        assert log_kwargs["body"] is None

tests/unit/aio/data/test_async_batch_internal.py

@@ -3,7 +3,6 @@
 
 """Unit tests for _AsyncBatchClient internals."""
 
-import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest