Sync async implementation with GA API: retrieve, list, expand/annotations support

Abel Milash · claude · Abel Milash · commit b84832407763 · 2026-05-11T16:53:43.000-07:00
- Add retrieve() and list() and list_pages() to AsyncRecordOperations (GA
  replacements for deprecated get())
- Add expand and include_annotations parameters to _AsyncODataClient._get()
  and _build_get() to match sync _ODataClient
- Add _build_list() to _AsyncODataClient for batch list support
- Add _RecordList handling to _AsyncBatchClient._resolve_item() and
  _resolve_record_list(); fix _resolve_record_get to pass expand and
  include_annotations
- Update test for _resolve_record_get to match new call signature

Co-Authored-By: Claude Sonnet 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/src/PowerPlatform/Dataverse/aio/data/_async_batch.py b/src/PowerPlatform/Dataverse/aio/data/_async_batch.py
@@ -18,6 +18,7 @@
     _RecordUpdate,
     _RecordDelete,
     _RecordGet,
+    _RecordList,
     _RecordUpsert,
     _TableCreate,
     _TableDelete,
@@ -180,6 +181,8 @@ async def _resolve_item(self, item: Any) -> List[_RawRequest]:
             return await self._resolve_record_delete(item)
         if isinstance(item, _RecordGet):
             return await self._resolve_record_get(item)
+        if isinstance(item, _RecordList):
+            return await self._resolve_record_list(item)
         if isinstance(item, _RecordUpsert):
             return await self._resolve_record_upsert(item)
         if isinstance(item, _TableCreate):
@@ -253,7 +256,30 @@ async def _resolve_record_delete(self, op: _RecordDelete) -> List[_RawRequest]:
         return requests
 
     async def _resolve_record_get(self, op: _RecordGet) -> List[_RawRequest]:
-        return [await self._od._build_get(op.table, op.record_id, select=op.select)]
+        return [
+            await self._od._build_get(
+                op.table,
+                op.record_id,
+                select=op.select,
+                expand=op.expand,
+                include_annotations=op.include_annotations,
+            )
+        ]
+
+    async def _resolve_record_list(self, op: _RecordList) -> List[_RawRequest]:
+        return [
+            await self._od._build_list(
+                op.table,
+                select=op.select,
+                filter=op.filter,
+                orderby=op.orderby,
+                top=op.top,
+                expand=op.expand,
+                page_size=op.page_size,
+                count=op.count,
+                include_annotations=op.include_annotations,
+            )
+        ]
 
     async def _resolve_record_upsert(self, op: _RecordUpsert) -> List[_RawRequest]:
         entity_set = await self._od._entity_set_from_schema_name(op.table)
diff --git a/src/PowerPlatform/Dataverse/aio/data/_async_odata.py b/src/PowerPlatform/Dataverse/aio/data/_async_odata.py
@@ -533,6 +533,8 @@ async def _get(
         table_schema_name: str,
         key: str,
         select: Optional[List[str]] = None,
+        expand: Optional[List[str]] = None,
+        include_annotations: Optional[str] = None,
     ) -> Dict[str, Any]:
         """Retrieve a single record.
 
@@ -542,11 +544,17 @@ async def _get(
         :type key: ``str``
         :param select: Columns to select; joined with commas into $select.
         :type select: ``list[str]`` | ``None``
+        :param expand: Navigation properties to expand (``$expand``); passed as-is (case-sensitive).
+        :type expand: ``list[str]`` | ``None``
+        :param include_annotations: OData annotation pattern for the ``Prefer: odata.include-annotations`` header, or ``None``.
+        :type include_annotations: ``str`` | ``None``
 
         :return: Retrieved record dictionary (may be empty if no selected attributes).
         :rtype: ``dict[str, Any]``
         """
-        r = await self._execute_raw(await self._build_get(table_schema_name, key, select=select))
+        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)
 
     async def _get_multiple(
@@ -1748,13 +1756,64 @@ async def _build_get(
         record_id: str,
         *,
         select: Optional[List[str]] = None,
+        expand: Optional[List[str]] = None,
+        include_annotations: Optional[str] = None,
     ) -> _RawRequest:
         """Build a single-record GET request without sending it."""
         entity_set = await self._entity_set_from_schema_name(table)
+        params: List[str] = []
+        if select:
+            params.append("$select=" + ",".join(self._lowercase_list(select)))
+        if expand:
+            params.append("$expand=" + ",".join(expand))
         url = f"{self.api}/{entity_set}{self._format_key(record_id)}"
+        if params:
+            url += "?" + "&".join(params)
+        headers = None
+        if include_annotations:
+            headers = {"Prefer": f'odata.include-annotations="{include_annotations}"'}
+        return _RawRequest(method="GET", url=url, headers=headers)
+
+    async def _build_list(
+        self,
+        table: str,
+        *,
+        select: Optional[List[str]] = None,
+        filter: Optional[str] = None,
+        orderby: Optional[List[str]] = None,
+        top: Optional[int] = None,
+        expand: Optional[List[str]] = None,
+        page_size: Optional[int] = None,
+        count: bool = False,
+        include_annotations: Optional[str] = None,
+    ) -> _RawRequest:
+        """Build a multi-record GET request (single page, no pagination) without sending it."""
+        entity_set = await self._entity_set_from_schema_name(table)
+        params: List[str] = []
         if select:
-            url += "?$select=" + ",".join(self._lowercase_list(select))
-        return _RawRequest(method="GET", url=url)
+            params.append("$select=" + ",".join(self._lowercase_list(select)))
+        if filter:
+            params.append("$filter=" + filter)
+        if orderby:
+            params.append("$orderby=" + ",".join(orderby))
+        if top is not None:
+            params.append(f"$top={top}")
+        if expand:
+            params.append("$expand=" + ",".join(expand))
+        if count:
+            params.append("$count=true")
+        url = f"{self.api}/{entity_set}"
+        if params:
+            url += "?" + "&".join(params)
+        prefer_parts: List[str] = []
+        if page_size is not None:
+            ps = int(page_size)
+            if ps > 0:
+                prefer_parts.append(f"odata.maxpagesize={ps}")
+        if include_annotations:
+            prefer_parts.append(f'odata.include-annotations="{include_annotations}"')
+        headers = {"Prefer": ",".join(prefer_parts)} if prefer_parts else None
+        return _RawRequest(method="GET", url=url, headers=headers)
 
     async def _build_sql(self, sql: str) -> _RawRequest:
         """Build a SQL query GET request without sending it.
diff --git a/src/PowerPlatform/Dataverse/aio/operations/async_records.py b/src/PowerPlatform/Dataverse/aio/operations/async_records.py
@@ -5,9 +5,10 @@
 
 from __future__ import annotations
 
-from typing import Any, AsyncIterator, Dict, List, Optional, Union, overload, TYPE_CHECKING
+from typing import Any, AsyncIterator, Dict, Iterator, List, Optional, Union, overload, TYPE_CHECKING
 
-from ...models.record import Record
+from ...core.errors import HttpError
+from ...models.record import QueryResult, Record
 from ...models.upsert import UpsertItem
 
 if TYPE_CHECKING:
@@ -463,6 +464,200 @@ async def _paged() -> AsyncIterator[List[Record]]:
 
         return _paged()
 
+    # --------------------------------------------------------------- retrieve
+
+    async def retrieve(
+        self,
+        table: str,
+        record_id: str,
+        *,
+        select: Optional[List[str]] = None,
+        expand: Optional[List[str]] = None,
+        include_annotations: Optional[str] = None,
+    ) -> Optional[Record]:
+        """Fetch a single record by its GUID, returning ``None`` if not found.
+
+        GA replacement for ``records.get(table, record_id)``. Returns ``None``
+        instead of raising when the record does not exist (HTTP 404).
+
+        :param table: Schema name of the table (e.g. ``"account"``).
+        :type table: :class:`str`
+        :param record_id: GUID of the record to retrieve.
+        :type record_id: :class:`str`
+        :param select: Optional list of column logical names to include.
+        :type select: list[str] or None
+        :param expand: Optional list of navigation properties to expand (e.g.
+            ``["primarycontactid"]``). Navigation property names are
+            case-sensitive and must match the entity's ``$metadata``.
+        :type expand: list[str] or None
+        :param include_annotations: OData annotation pattern for the
+            ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or
+            ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
+        :type include_annotations: :class:`str` or None
+        :return: Typed record, or ``None`` if not found.
+        :rtype: :class:`~PowerPlatform.Dataverse.models.record.Record` or None
+
+        Example::
+
+            record = await client.records.retrieve(
+                "account", account_id,
+                select=["name", "statuscode"],
+                expand=["primarycontactid"],
+                include_annotations="OData.Community.Display.V1.FormattedValue",
+            )
+            if record is not None:
+                contact = record.get("primarycontactid") or {}
+                print(contact.get("fullname"))
+        """
+        async with self._client._scoped_odata() as od:
+            try:
+                raw = await od._get(table, record_id, select=select, expand=expand, include_annotations=include_annotations)
+            except HttpError as exc:
+                if exc.status_code == 404:
+                    return None
+                raise
+            return Record.from_api_response(table, raw, record_id=record_id)
+
+    # -------------------------------------------------------------------- list
+
+    async def list(
+        self,
+        table: str,
+        *,
+        filter: Optional[Union[str, Any]] = None,
+        select: Optional[List[str]] = None,
+        orderby: Optional[List[str]] = None,
+        top: Optional[int] = None,
+        expand: Optional[List[str]] = None,
+        page_size: Optional[int] = None,
+        count: bool = False,
+        include_annotations: Optional[str] = None,
+    ) -> QueryResult:
+        """Fetch multiple records and return them as a :class:`QueryResult`.
+
+        GA replacement for ``records.get(table, filter=...)``. All pages are
+        collected eagerly and returned as a single :class:`QueryResult`.
+
+        :param table: Schema name of the table (e.g. ``"account"``).
+        :type table: :class:`str`
+        :param filter: Optional OData filter string or :class:`FilterExpression`.
+        :type filter: str or FilterExpression or None
+        :param select: Optional list of column logical names to include.
+        :type select: list[str] or None
+        :param orderby: Optional list of sort expressions (e.g. ``["name asc", "createdon desc"]``).
+        :type orderby: list[str] or None
+        :param top: Maximum total number of records to return.
+        :type top: int or None
+        :param expand: Optional list of navigation properties to expand.
+        :type expand: list[str] or None
+        :param page_size: Per-page size hint via ``Prefer: odata.maxpagesize``.
+        :type page_size: int or None
+        :param count: If ``True``, adds ``$count=true`` to include a total record count.
+        :type count: bool
+        :param include_annotations: OData annotation pattern for the
+            ``Prefer: odata.include-annotations`` header, or ``None``.
+        :type include_annotations: :class:`str` or None
+        :return: All matching records collected into a :class:`QueryResult`.
+        :rtype: :class:`~PowerPlatform.Dataverse.models.record.QueryResult`
+
+        Example::
+
+            result = await client.records.list(
+                "account",
+                filter="statecode eq 0",
+                select=["name", "statuscode"],
+                orderby=["name asc"],
+                top=100,
+                include_annotations="OData.Community.Display.V1.FormattedValue",
+            )
+            for record in result:
+                print(record["name"])
+        """
+        filter_str: Optional[str] = str(filter) if filter is not None else None
+        all_records: List[Record] = []
+        async with self._client._scoped_odata() as od:
+            async for page in od._get_multiple(
+                table,
+                select=select,
+                filter=filter_str,
+                orderby=orderby,
+                top=top,
+                expand=expand,
+                page_size=page_size,
+                count=count,
+                include_annotations=include_annotations,
+            ):
+                all_records.extend(Record.from_api_response(table, row) for row in page)
+        return QueryResult(all_records)
+
+    # --------------------------------------------------------------- list_pages
+
+    async def list_pages(
+        self,
+        table: str,
+        *,
+        filter: Optional[Union[str, Any]] = None,
+        select: Optional[List[str]] = None,
+        orderby: Optional[List[str]] = None,
+        top: Optional[int] = None,
+        expand: Optional[List[str]] = None,
+        page_size: Optional[int] = None,
+        count: bool = False,
+        include_annotations: Optional[str] = None,
+    ) -> AsyncIterator[QueryResult]:
+        """Lazily yield one :class:`QueryResult` per HTTP page.
+
+        Streaming counterpart to :meth:`list`. Each iteration triggers one
+        network request via ``@odata.nextLink``. One-shot — do not iterate
+        more than once.
+
+        :param table: Schema name of the table (e.g. ``"account"``).
+        :type table: :class:`str`
+        :param filter: Optional OData filter string or :class:`FilterExpression`.
+        :type filter: str or FilterExpression or None
+        :param select: Optional list of column logical names to include.
+        :type select: list[str] or None
+        :param orderby: Optional list of sort expressions.
+        :type orderby: list[str] or None
+        :param top: Maximum total number of records to return.
+        :type top: int or None
+        :param expand: Optional list of navigation properties to expand.
+        :type expand: list[str] or None
+        :param page_size: Per-page size hint via ``Prefer: odata.maxpagesize``.
+        :type page_size: int or None
+        :param count: If ``True``, adds ``$count=true`` to include a total record count.
+        :type count: bool
+        :param include_annotations: OData annotation pattern for the
+            ``Prefer: odata.include-annotations`` header, or ``None``.
+        :type include_annotations: :class:`str` or None
+        :return: Async iterator of per-page :class:`QueryResult` objects.
+        :rtype: AsyncIterator[:class:`~PowerPlatform.Dataverse.models.record.QueryResult`]
+
+        Example::
+
+            async for page in client.records.list_pages(
+                "account",
+                filter="statecode eq 0",
+                orderby=["name asc"],
+                page_size=200,
+            ):
+                process(page.to_dataframe())
+        """
+        filter_str: Optional[str] = str(filter) if filter is not None else None
+        async with self._client._scoped_odata() as od:
+            async for page in od._get_multiple(
+                table,
+                select=select,
+                filter=filter_str,
+                orderby=orderby,
+                top=top,
+                expand=expand,
+                page_size=page_size,
+                count=count,
+                include_annotations=include_annotations,
+            ):
+                yield QueryResult([Record.from_api_response(table, row) for row in page])
+
     # ------------------------------------------------------------------ upsert
 
     async def upsert(self, table: str, items: List[Union[UpsertItem, Dict[str, Any]]]) -> None:
diff --git a/tests/unit/aio/data/test_async_batch_internal.py b/tests/unit/aio/data/test_async_batch_internal.py
@@ -268,7 +268,7 @@ async def test_single_get_request(self):
         op = _RecordGet(table="account", record_id="guid-1", select=["name"])
         result = await client._resolve_record_get(op)
         assert len(result) == 1
-        od._build_get.assert_called_once_with("account", "guid-1", select=["name"])
+        od._build_get.assert_called_once_with("account", "guid-1", select=["name"], expand=None, include_annotations=None)
 
 
 # ---------------------------------------------------------------------------
