Async review pass: align with GA API, fix async patterns, remove deprecated surface

- Remove deprecated filter_eq/filter_ne/filter_in methods from async_records and async_query (not in GA API)
- Remove stale tests for removed methods
- Use asyncio.gather() for independent I/O: relationship sub-requests, batch delete/remove-columns, multi-record label conversion, delete_columns metadata lookups
- Fix blocking file I/O in _async_upload: asyncio.to_thread() for reads, Path instead of os.path
- Remove lazy imports in _async_relationships; move to module level
- Fix test_async_batch: wrap deprecated records.get() call in pytest.warns(DeprecationWarning)
- Remove unused Optional import from _async_batch
- Remove redundant return None statements in _async_odata

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

Author: Abel Milash

src/PowerPlatform/Dataverse/aio/async_client.py

@@ -71,10 +71,10 @@ class AsyncDataverseClient:
     Example:
         **Recommended -- async context manager** (enables HTTP connection pooling)::
 
-            from azure.identity.aio import DefaultAzureCredential
+            from azure.identity.aio import InteractiveBrowserCredential
             from PowerPlatform.Dataverse.aio.async_client import AsyncDataverseClient
 
-            credential = DefaultAzureCredential()
+            credential = InteractiveBrowserCredential()
 
             async with AsyncDataverseClient("https://org.crm.dynamics.com", credential) as client:
                 record_id = await client.records.create("account", {"name": "Contoso Ltd"})
@@ -118,12 +118,17 @@ def _get_odata(self) -> _AsyncODataClient:
         Get or create the internal async OData client instance.
 
         This method implements lazy initialization of the low-level async OData
-        client, deferring construction until the first API call.
+        client, deferring construction until the first API call. When used outside
+        of an ``async with`` block, a :class:`aiohttp.ClientSession` is created
+        lazily here so that standalone usage (without a context manager) works
+        without requiring the caller to manage the session explicitly.
 
         :return: The lazily-initialized low-level async client.
         :rtype: ~PowerPlatform.Dataverse.aio.data._async_odata._AsyncODataClient
         """
         if self._odata is None:
+            if self._session is None:
+                self._session = aiohttp.ClientSession()
             self._odata = _AsyncODataClient(
                 self.auth,
                 self._base_url,
@@ -199,3 +204,32 @@ def _check_closed(self) -> None:
         """Raise :class:`RuntimeError` if the client has been closed."""
         if self._closed:
             raise RuntimeError("AsyncDataverseClient is closed")
+
+    # ---------------- Cache utilities ----------------
+
+    async def flush_cache(self, kind: str) -> int:
+        """
+        Flush cached client metadata or state.
+
+        :param kind: Cache kind to flush. Currently supported values:
+
+            - ``"picklist"``: Clears picklist label cache used for label-to-integer conversion
+
+            Future kinds (e.g. ``"entityset"``, ``"primaryid"``) may be added without
+            breaking this signature.
+        :type kind: :class:`str`
+
+        :return: Number of cache entries removed.
+        :rtype: :class:`int`
+
+        Example:
+            Clear the picklist cache::
+
+                removed = await client.flush_cache("picklist")
+                print(f"Cleared {removed} cached picklist entries")
+        """
+        async with self._scoped_odata() as od:
+            return od._flush_cache(kind)
+
+
+__all__ = ["AsyncDataverseClient"]

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

@@ -5,8 +5,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import uuid
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Union
 
 from ...core.errors import MetadataError, ValidationError
 from ...core._error_codes import METADATA_TABLE_NOT_FOUND, METADATA_COLUMN_NOT_FOUND
@@ -250,10 +251,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)]
-        requests: List[_RawRequest] = []
-        for rid in ids:
-            requests.append(await self._od._build_delete(op.table, rid))
-        return requests
+        return list(await asyncio.gather(*[self._od._build_delete(op.table, rid) for rid in ids]))
 
     async def _resolve_record_get(self, op: _RecordGet) -> List[_RawRequest]:
         return [
@@ -317,11 +315,12 @@ 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")
+            for col_name in columns
+        ])
         requests: List[_RawRequest] = []
-        for col_name in columns:
-            attr_meta = await self._od._get_attribute_metadata(
-                metadata_id, col_name, extra_select="@odata.type,AttributeType"
-            )
+        for col_name, attr_meta in zip(columns, attr_metas):
             if not attr_meta or not attr_meta.get("MetadataId"):
                 raise MetadataError(
                     f"Column '{col_name}' not found on table '{op.table}'.",

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

@@ -367,11 +367,13 @@ async def _upsert_multiple(
                 f"alternate_keys and records must have the same length " f"({len(alternate_keys)} != {len(records)})"
             )
         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
+        ])
         targets: List[Dict[str, Any]] = []
-        for alt_key, record in zip(alternate_keys, records):
+        for alt_key, record_processed in zip(alternate_keys, converted):
             alt_key_lower = self._lowercase_keys(alt_key)
-            record_processed = self._lowercase_keys(record)
-            record_processed = await self._convert_labels_to_ints(table_schema_name, record_processed)
             conflicting = {
                 k for k in set(alt_key_lower) & set(record_processed) if alt_key_lower[k] != record_processed[k]
             }
@@ -429,7 +431,7 @@ async def _update_by_ids(
         if isinstance(changes, dict):
             batch = [{pk_attr: rid, **changes} for rid in ids]
             await self._update_multiple(entity_set, table_schema_name, batch)
-            return None
+            return
         if not isinstance(changes, list):
             raise TypeError("changes must be dict or list[dict]")
         if len(changes) != len(ids):
@@ -440,7 +442,6 @@ async def _update_by_ids(
                 raise TypeError("Each patch must be a dict")
             batch.append({pk_attr: rid, **patch})
         await self._update_multiple(entity_set, table_schema_name, batch)
-        return None
 
     async def _delete_multiple(
         self,
@@ -513,7 +514,6 @@ async def _update_multiple(
         if not isinstance(records, list) or not records or not all(isinstance(r, dict) for r in records):
             raise TypeError("records must be a non-empty list[dict]")
         await self._execute_raw(await self._build_update_multiple_from_records(entity_set, table_schema_name, records))
-        return None
 
     async def _delete(self, table_schema_name: str, key: str) -> None:
         """Delete a record by GUID.
@@ -1477,10 +1477,11 @@ async def _delete_columns(
         deleted: List[str] = []
         needs_picklist_flush = False
 
-        for column_name in names:
-            attr_meta = await self._get_attribute_metadata(
-                metadata_id, column_name, extra_select="@odata.type,AttributeType"
-            )
+        attr_metas = await asyncio.gather(*[
+            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(
                     f"Column '{column_name}' not found on table '{entity_schema}'.",
@@ -1536,13 +1537,12 @@ async def _build_create_multiple(
         if not all(isinstance(r, dict) for r in records):
             raise TypeError("All items for multi-create must be dicts")
         logical_name = table.lower()
-        enriched = []
-        for r in records:
-            r = self._lowercase_keys(r)
-            r = await self._convert_labels_to_ints(table, r)
-            if "@odata.type" not in r:
-                r = {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"}
-            enriched.append(r)
+        lowered = [self._lowercase_keys(r) for r in records]
+        converted = await asyncio.gather(*[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
+        ]
         return _RawRequest(
             method="POST",
             url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.CreateMultiple",
@@ -1590,13 +1590,12 @@ async def _build_update_multiple_from_records(
         :meth:`_build_update_multiple` (which assembles from ids + changes).
         """
         logical_name = table.lower()
-        enriched = []
-        for r in records:
-            r = self._lowercase_keys(r)
-            r = await self._convert_labels_to_ints(table, r)
-            if "@odata.type" not in r:
-                r = {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"}
-            enriched.append(r)
+        lowered = [self._lowercase_keys(r) for r in records]
+        converted = await asyncio.gather(*[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
+        ]
         return _RawRequest(
             method="POST",
             url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpdateMultiple",
@@ -1661,11 +1660,11 @@ async def _build_upsert_multiple(
                 subcode="upsert_length_mismatch",
             )
         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])
         targets: List[Dict[str, Any]] = []
-        for alt_key, record in zip(alternate_keys, records):
+        for alt_key, record_processed in zip(alternate_keys, converted):
             alt_key_lower = self._lowercase_keys(alt_key)
-            record_processed = self._lowercase_keys(record)
-            record_processed = await self._convert_labels_to_ints(table, record_processed)
             conflicting = {
                 k for k in set(alt_key_lower) & set(record_processed) if alt_key_lower[k] != record_processed[k]
             }

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

@@ -11,9 +11,13 @@
 
 __all__ = []
 
+import asyncio
 import re
 from typing import Any, Dict, List, Optional
 
+from ...core.errors import MetadataError
+from ...core._error_codes import METADATA_TABLE_NOT_FOUND
+
 
 class _AsyncRelationshipOperationsMixin:
     """
@@ -207,9 +211,6 @@ async def _list_table_relationships(
         :raises MetadataError: If the table is not found.
         :raises HttpError: If the Web API request fails.
         """
-        from ...core.errors import MetadataError
-        from ...core._error_codes import METADATA_TABLE_NOT_FOUND
-
         ent = await self._get_entity_by_table_schema_name(table_schema_name)
         if not ent or not ent.get("MetadataId"):
             raise MetadataError(
@@ -237,9 +238,12 @@ async def _list_table_relationships(
         many_to_one_url = f"{self.api}/EntityDefinitions({metadata_id})/ManyToOneRelationships"
         many_to_many_url = f"{self.api}/EntityDefinitions({metadata_id})/ManyToManyRelationships"
 
-        r1 = await self._request("get", one_to_many_url, headers=await self._headers(), params=one_to_many_params)
-        r2 = await self._request("get", many_to_one_url, headers=await self._headers(), params=one_to_many_params)
-        r3 = await self._request("get", many_to_many_url, headers=await self._headers(), params=many_to_many_params)
+        headers = await self._headers()
+        r1, r2, r3 = await asyncio.gather(
+            self._request("get", one_to_many_url, headers=headers, params=one_to_many_params),
+            self._request("get", many_to_one_url, headers=headers, params=one_to_many_params),
+            self._request("get", many_to_many_url, headers=headers, params=many_to_many_params),
+        )
 
         return (
             (await r1.json()).get("value", []) + (await r2.json()).get("value", []) + (await r3.json()).get("value", [])

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

@@ -5,7 +5,10 @@
 
 from __future__ import annotations
 
+import asyncio
+from pathlib import Path
 from typing import Optional
+from urllib.parse import quote
 
 
 class _AsyncFileUploadMixin:
@@ -40,8 +43,6 @@ async def _upload_file(
         if_none_match : :class:`bool`
             When True (default) only succeeds if column empty. When False overwrites (If-Match: *).
         """
-        import os
-
         # Resolve entity set from table schema name
         entity_set = await self._entity_set_from_schema_name(table_schema_name)
 
@@ -61,9 +62,10 @@ async def _upload_file(
         mode = (mode or "auto").lower()
 
         if mode == "auto":
-            if not os.path.isfile(path):
+            p = Path(path)
+            if not p.is_file():
                 raise FileNotFoundError(f"File not found: {path}")
-            size = os.path.getsize(path)
+            size = p.stat().st_size
             mode = "small" if size < 128 * 1024 * 1024 else "chunk"
 
         # Convert schema name to lowercase logical name for URL usage
@@ -87,19 +89,17 @@ async def _upload_file_small(
         if_none_match: bool = True,
     ) -> None:
         """Upload a file (<128MB) via single PATCH."""
-        import os
-
         if not record_id:
             raise ValueError("record_id required")
-        if not os.path.isfile(path):
+        p = Path(path)
+        if not p.is_file():
             raise FileNotFoundError(f"File not found: {path}")
-        size = os.path.getsize(path)
+        size = p.stat().st_size
         limit = 128 * 1024 * 1024
         if size > limit:
             raise ValueError(f"File size {size} exceeds single-upload limit {limit}; use chunk mode.")
-        with open(path, "rb") as fh:
-            data = fh.read()
-        fname = os.path.basename(path)
+        data = await asyncio.to_thread(p.read_bytes)
+        fname = p.name
         key = self._format_key(record_id)
         url = f"{self.api}/{entity_set}{key}/{file_name_attribute}"
         headers = {
@@ -112,7 +112,6 @@ async def _upload_file_small(
             headers["If-Match"] = "*"
         # Single PATCH upload; allow default success codes (includes 204)
         await self._request("patch", url, headers=headers, data=data)
-        return None
 
     async def _upload_file_chunk(
         self,
@@ -139,20 +138,19 @@ async def _upload_file_chunk(
         if_none_match : :class:`bool`
             When True sends ``If-None-Match: null`` to only succeed if the column is currently empty.
             Set False to always overwrite (uses ``If-Match: *``).
+
         Returns
         -------
         None
             Returns nothing on success. Any failure raises an exception.
         """
-        import os, math
-        from urllib.parse import quote
-
         if not record_id:
             raise ValueError("record_id required")
-        if not os.path.isfile(path):
+        p = Path(path)
+        if not p.is_file():
             raise FileNotFoundError(f"File not found: {path}")
-        total_size = os.path.getsize(path)
-        fname = os.path.basename(path)
+        total_size = p.stat().st_size
+        fname = p.name
         key = self._format_key(record_id)
         init_url = f"{self.api}/{entity_set}{key}/{file_name_attribute}?x-ms-file-name={quote(fname)}"
         headers = {
@@ -174,11 +172,11 @@ async def _upload_file_chunk(
         effective_size = recommended_size or (4 * 1024 * 1024)
         if effective_size <= 0:
             raise ValueError("effective chunk size must be positive")
-        total_chunks = int(math.ceil(total_size / effective_size)) if total_size else 1
+        total_chunks = (total_size + effective_size - 1) // effective_size if total_size else 1
         uploaded_bytes = 0
-        with open(path, "rb") as fh:
-            for idx in range(total_chunks):
-                chunk = fh.read(effective_size)
+        with p.open("rb") as fh:
+            for _ in range(total_chunks):
+                chunk = await asyncio.to_thread(fh.read, effective_size)
                 if not chunk:
                     break
                 start = uploaded_bytes
@@ -192,4 +190,3 @@ async def _upload_file_chunk(
                 # Each chunk returns 206 (partial) or 204 (final). Accept both.
                 await self._request("patch", location, headers=c_headers, data=chunk, expected=(206, 204))
                 uploaded_bytes += len(chunk)
-        return None