Replace _SyncResponseWrapper with _HttpResponse in async HTTP layer

Adds _HttpResponse to _async_http.py — a materialized response that
buffers the body bytes in _request, exposing sync .text and .json()
so callers need no await. Eliminates _SyncResponseWrapper from
_async_batch.py and removes all remaining await r.json() calls
throughout the async layer (_async_odata.py, _async_relationships.py).
Also removes aiohttp.ContentTypeError from except clauses now that
json() is a plain json.loads call (ValueError covers it).

All 2155 tests pass.

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

Author: Abel Milash

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

@@ -12,15 +12,45 @@
 from __future__ import annotations
 
 import asyncio
+import json as _json
 import time
-from typing import TYPE_CHECKING, Any, Optional
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
 import aiohttp
 
 if TYPE_CHECKING:
     from ...core._http_logger import _HttpLogger
 
 
+class _HttpResponse:
+    """Materialized HTTP response returned by :class:`_AsyncHttpClient._request`.
+
+    The body is fully buffered before this object is constructed, so all
+    accessors are synchronous — no ``await`` required.
+
+    :param status: HTTP status code.
+    :param headers: Response headers as a plain dict.
+    :param body: Raw response body bytes.
+    """
+
+    __slots__ = ("status", "status_code", "headers", "_body")
+
+    def __init__(self, status: int, headers: Dict[str, str], body: bytes) -> None:
+        self.status = status
+        self.status_code = status  # alias used by batch response parser
+        self.headers = headers
+        self._body = body
+
+    @property
+    def text(self) -> str:
+        """Response body decoded as UTF-8 text."""
+        return self._body.decode("utf-8", errors="replace") if self._body else ""
+
+    def json(self, content_type: Any = None) -> Any:
+        """Parse and return the response body as JSON."""
+        return _json.loads(self._body) if self._body else {}
+
+
 class _AsyncHttpClient:
     """
     Async HTTP client with configurable retry logic and timeout handling.
@@ -58,24 +88,24 @@ def __init__(
         self._session = session
         self._logger = logger
 
-    async def _request(self, method: str, url: str, **kwargs: Any) -> aiohttp.ClientResponse:
+    async def _request(self, method: str, url: str, **kwargs: Any) -> _HttpResponse:
         """
         Execute an HTTP request asynchronously with automatic retry logic and timeout management.
 
         Applies default timeouts based on HTTP method (120s for POST/DELETE, 10s for others)
         and retries on network errors with exponential backoff.
 
-        The response body is fully buffered before returning, so the caller may freely
-        call ``await resp.text()`` or ``await resp.json()`` without managing the connection lifecycle.
+        The response body is fully buffered and returned as a :class:`_HttpResponse` whose
+        accessors (``.text``, ``.json()``) are synchronous — no ``await`` required on the caller side.
 
         :param method: HTTP method (GET, POST, PUT, DELETE, etc.).
         :type method: :class:`str`
         :param url: Target URL for the request.
         :type url: :class:`str`
         :param kwargs: Additional arguments passed to ``aiohttp.ClientSession.request()``,
             including headers, data, etc.
-        :return: HTTP response object with body fully buffered.
-        :rtype: :class:`aiohttp.ClientResponse`
+        :return: Materialized HTTP response with body fully buffered.
+        :rtype: :class:`_HttpResponse`
         :raises aiohttp.ClientError: If all retry attempts fail.
         :raises RuntimeError: If no session has been set.
         """
@@ -113,27 +143,23 @@ async def _request(self, method: str, url: str, **kwargs: Any) -> aiohttp.Client
             try:
                 t0 = time.monotonic()
                 async with self._session.request(method, url, **kwargs) as resp:
-                    await resp.read()
+                    body = await resp.read()
+                    materialized = _HttpResponse(resp.status, dict(resp.headers), body)
                 elapsed_ms = (time.monotonic() - t0) * 1000
 
                 if self._logger is not None:
-                    # Only decode resp.text when body logging is enabled — avoids
+                    # Only decode text when body logging is enabled — avoids
                     # unnecessary overhead for large payloads when max_body_bytes == 0.
-                    resp_body = None
-                    if self._logger.body_logging_enabled:
-                        try:
-                            resp_body = await resp.text()
-                        except Exception:
-                            pass
+                    resp_body = materialized.text if self._logger.body_logging_enabled else None
                     self._logger.log_response(
                         method,
                         url,
-                        status_code=resp.status,
-                        headers=dict(resp.headers),
+                        status_code=materialized.status,
+                        headers=materialized.headers,
                         body=resp_body,
                         elapsed_ms=elapsed_ms,
                     )
-                return resp
+                return materialized
             except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                 if self._logger is not None:
                     self._logger.log_error(

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

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 import uuid
-from typing import TYPE_CHECKING, Any, Dict, List, Union
+from typing import TYPE_CHECKING, Any, List, Union
 
 from ...core.errors import MetadataError, ValidationError
 from ...core._error_codes import METADATA_TABLE_NOT_FOUND, METADATA_COLUMN_NOT_FOUND
@@ -43,32 +43,6 @@
 __all__ = []
 
 
-class _SyncResponseWrapper:
-    """Minimal requests-compatible wrapper around a materialized aiohttp response.
-
-    Used to feed the sync-only :meth:`_BatchBase._parse_batch_response` helper with
-    pre-fetched body content, avoiding the need to ``await`` inside sync code.
-
-    :param status_code: HTTP status code (e.g. 200, 400).
-    :type status_code: :class:`int`
-    :param headers: Response headers as a plain dict.
-    :type headers: :class:`dict`
-    :param text: Full response body as a decoded string.
-    :type text: :class:`str`
-    :param json_payload: Pre-parsed JSON body (or empty dict if parsing failed).
-    :type json_payload: ``Any``
-    """
-
-    def __init__(self, status_code: int, headers: Dict[str, str], text: str, json_payload: Any) -> None:
-        self.status_code = status_code
-        self.headers = headers
-        self.text = text
-        self._json = json_payload
-
-    def json(self) -> Any:
-        return self._json
-
-
 # ---------------------------------------------------------------------------
 # Batch client: resolves intents → raw requests → multipart body → HTTP → result
 # ---------------------------------------------------------------------------
@@ -116,7 +90,7 @@ async def execute(
         batch_boundary = f"batch_{uuid.uuid4()}"
         body = self._build_batch_body(resolved, batch_boundary)
 
-        headers: Dict[str, str] = {
+        headers: dict[str, str] = {
             "Content-Type": f'multipart/mixed; boundary="{batch_boundary}"',
         }
         if continue_on_error:
@@ -135,22 +109,7 @@ async def execute(
             expected=(200, 202, 207, 400),
         )
 
-        # Materialise the response body so the sync _parse_batch_response can
-        # access it without awaiting.  Body is already buffered (await resp.read()
-        # in _AsyncHttpClient._request), so these calls are fast and non-blocking.
-        text = await r.text()
-        try:
-            json_payload = await r.json(content_type=None)
-        except Exception:
-            json_payload = {}
-
-        wrapped = _SyncResponseWrapper(
-            status_code=r.status,
-            headers=dict(r.headers),
-            text=text,
-            json_payload=json_payload,
-        )
-        return self._parse_batch_response(wrapped)
+        return self._parse_batch_response(r)
 
     # ------------------------------------------------------------------
     # Intent resolution dispatcher

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

@@ -16,9 +16,7 @@
 
 from urllib.parse import quote as _url_quote
 
-import aiohttp
-
-from ..core._async_http import _AsyncHttpClient
+from ..core._async_http import _AsyncHttpClient, _HttpResponse
 from ._async_upload import _AsyncFileUploadMixin
 from ._async_relationships import _AsyncRelationshipOperationsMixin
 from ...core.errors import *
@@ -109,7 +107,7 @@ async def _headers(self) -> Dict[str, str]:
             "User-Agent": ua,
         }
 
-    async def _raw_request(self, method: str, url: str, **kwargs) -> aiohttp.ClientResponse:
+    async def _raw_request(self, method: str, url: str, **kwargs) -> _HttpResponse:
         return await self._http._request(method, url, **kwargs)
 
     async def _request(
@@ -119,7 +117,7 @@ async def _request(
         *,
         expected: tuple[int, ...] = _DEFAULT_EXPECTED_STATUSES,
         **kwargs,
-    ) -> aiohttp.ClientResponse:
+    ) -> _HttpResponse:
         # Acquire base headers once (async), then use a sync closure for _RequestContext.build.
         # _RequestContext.build is a sync classmethod defined in _ODataBase and shared by both
         # sync and async clients — keeping it sync avoids duplicating the header-injection logic.
@@ -145,11 +143,7 @@ def _merge(h: Optional[Dict[str, str]]) -> Dict[str, str]:
             return r
 
         response_headers = getattr(r, "headers", {}) or {}
-        raw_text = ""
-        try:
-            raw_text = await r.text()
-        except Exception:
-            pass
+        raw_text = r.text
         body_excerpt = raw_text[:200]
         svc_code = None
         msg = f"HTTP {r.status}"
@@ -207,7 +201,7 @@ async def _execute_raw(
         req: _RawRequest,
         *,
         expected: tuple[int, ...] = _DEFAULT_EXPECTED_STATUSES,
-    ) -> aiohttp.ClientResponse:
+    ) -> _HttpResponse:
         """Execute a ``_RawRequest`` and return the HTTP response.
 
         Encodes the pre-serialised body (if present) as UTF-8 and merges any
@@ -278,8 +272,8 @@ async def _create_multiple(
             raise TypeError("All items for multi-create must be dicts")
         r = await self._execute_raw(await self._build_create_multiple(entity_set, table_schema_name, records))
         try:
-            body = await r.json(content_type=None)
-        except (ValueError, aiohttp.ContentTypeError):
+            body = r.json()
+        except ValueError:
             body = {}
         if not isinstance(body, dict):
             return []
@@ -468,8 +462,8 @@ async def _delete_multiple(
         )
         job_id = None
         try:
-            body = await response.json(content_type=None)
-        except (ValueError, aiohttp.ContentTypeError):
+            body = response.json()
+        except ValueError:
             body = {}
         if isinstance(body, dict):
             job_id = body.get("JobId")
@@ -556,7 +550,7 @@ async def _get(
         r = await self._execute_raw(
             await self._build_get(table_schema_name, key, select=select, expand=expand, include_annotations=include_annotations)
         )
-        return await r.json(content_type=None)
+        return r.json()
 
     async def _get_multiple(
         self,
@@ -609,8 +603,8 @@ async def _do_request(url: str, *, params: Optional[Dict[str, Any]] = None) -> D
             headers = extra_headers if extra_headers else None
             r = await self._request("get", url, headers=headers, params=params)
             try:
-                return await r.json(content_type=None)
-            except (ValueError, aiohttp.ContentTypeError):
+                return r.json()
+            except ValueError:
                 return {}
 
         entity_set = await self._entity_set_from_schema_name(table_schema_name)
@@ -682,8 +676,8 @@ async def _query_sql(self, sql: str) -> list[dict[str, Any]]:
 
         r = await self._execute_raw(await self._build_sql(sql))
         try:
-            body = await r.json(content_type=None)
-        except (ValueError, aiohttp.ContentTypeError):
+            body = r.json()
+        except ValueError:
             return []
 
         # Collect first page
@@ -745,8 +739,8 @@ async def _query_sql(self, sql: str) -> list[dict[str, Any]]:
                 )
                 break
             try:
-                page_body = await page_resp.json(content_type=None)
-            except (ValueError, aiohttp.ContentTypeError) as exc:
+                page_body = page_resp.json()
+            except ValueError as exc:
                 warnings.warn(
                     f"SQL pagination stopped after {len(results)} rows — "
                     f"the next-page response was not valid JSON: {exc}. "
@@ -790,9 +784,9 @@ async def _entity_set_from_schema_name(self, table_schema_name: str) -> str:
         }
         r = await self._request("get", url, params=params)
         try:
-            body = await r.json(content_type=None)
+            body = r.json()
             items = body.get("value", []) if isinstance(body, dict) else []
-        except (ValueError, aiohttp.ContentTypeError):
+        except ValueError:
             items = []
         if not items:
             plural_hint = (
@@ -838,7 +832,7 @@ async def _get_entity_by_table_schema_name(
             "$filter": f"LogicalName eq '{logical_escaped}'",
         }
         r = await self._request("get", url, params=params, headers=headers)
-        items = (await r.json(content_type=None)).get("value", [])
+        items = (r.json()).get("value", [])
         return items[0] if items else None
 
     async def _create_entity(
@@ -903,8 +897,8 @@ async def _get_attribute_metadata(
         }
         r = await self._request("get", url, params=params)
         try:
-            body = await r.json(content_type=None)
-        except (ValueError, aiohttp.ContentTypeError):
+            body = r.json()
+        except ValueError:
             return None
         items = body.get("value") if isinstance(body, dict) else None
         if isinstance(items, list) and items:
@@ -955,7 +949,7 @@ async def _list_columns(
         if filter:
             params["$filter"] = filter
         r = await self._request("get", url, params=params)
-        return (await r.json(content_type=None)).get("value", [])
+        return (r.json()).get("value", [])
 
     async def _wait_for_attribute_visibility(
         self,
@@ -991,7 +985,7 @@ async def _wait_for_attribute_visibility(
             f"after {total_wait} seconds (exhausted all retries)."
         ) from last_error
 
-    async def _request_metadata_with_retry(self, method: str, url: str, **kwargs) -> aiohttp.ClientResponse:
+    async def _request_metadata_with_retry(self, method: str, url: str, **kwargs) -> _HttpResponse:
         """Fetch metadata with retries on transient errors."""
         max_attempts = 5
         backoff_seconds = 0.4
@@ -1038,7 +1032,7 @@ async def _bulk_fetch_picklists(self, table_schema_name: str) -> None:
                 f"?$select=LogicalName&$expand=OptionSet($select=Options)"
             )
             response = await self._request_metadata_with_retry("get", url)
-            body = await response.json(content_type=None)
+            body = response.json()
             items = body.get("value", []) if isinstance(body, dict) else []
 
             picklists: Dict[str, Dict[str, int]] = {}
@@ -1165,7 +1159,7 @@ async def _list_tables(
         :raises HttpError: If the metadata request fails.
         """
         r = await self._execute_raw(self._build_list_entities(filter=filter, select=select))
-        return (await r.json(content_type=None)).get("value", [])
+        return (r.json()).get("value", [])
 
     async def _delete_table(self, table_schema_name: str) -> None:
         """Delete a table by schema name.
@@ -1264,7 +1258,7 @@ async def _get_alternate_keys(self, table_schema_name: str) -> List[Dict[str, An
         logical_name = ent.get("LogicalName", table_schema_name.lower())
         url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys"
         r = await self._request("get", url)
-        return (await r.json(content_type=None)).get("value", [])
+        return (r.json()).get("value", [])
 
     async def _delete_alternate_key(self, table_schema_name: str, key_id: str) -> None:
         """Delete an alternate key by metadata ID.