Fix 6 bug bash bugs: SQL TOP enforcement, QueryResult __aiter__, token caching/coalescing, auth exception sanitization, empty data validation, GUID validation

Bug 1 (ADO 6431970): Client-side enforcement of SELECT TOP n for ?sql= endpoint
Bug 2 (ADO 6431066): Add __aiter__ to QueryResult for async for support
Bug 3 (ADO 6431085): Token caching + lock-based coalescing in auth managers
Bug 4 (ADO 6431086): Wrap credential exceptions in AuthenticationError to prevent token leaks
Bug 5 (ADO 6431942): Reject empty data in records.create()
Bug 6 (ADO 6431966): Validate GUIDs in retrieve/update/delete before server call

All 2431 unit tests pass.

Author: Abel Milash

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

@@ -12,9 +12,16 @@
 
 from __future__ import annotations
 
+import asyncio
+import random
+import time
+from typing import Optional
+
+from azure.core.credentials import AccessToken
 from azure.core.credentials_async import AsyncTokenCredential
 
-from ...core._auth import _TokenPair
+from ...core._auth import _TokenPair, _REFRESH_WINDOW_SECONDS, _MAX_REFRESH_JITTER_SECONDS
+from ...core.errors import AuthenticationError
 
 
 class _AsyncAuthManager:
@@ -30,16 +37,47 @@ def __init__(self, credential: AsyncTokenCredential) -> None:
         if not isinstance(credential, AsyncTokenCredential):
             raise TypeError("credential must implement azure.core.credentials_async.AsyncTokenCredential.")
         self.credential: AsyncTokenCredential = credential
+        self._token: Optional[AccessToken] = None
+        self._lock = asyncio.Lock()
+        self._refresh_jitter: int = 0
+
+    def _need_new_token(self) -> bool:
+        """Return ``True`` when the cached token is missing or near expiry."""
+        if self._token is None:
+            return True
+        remaining = self._token.expires_on - time.time()
+        if remaining <= 0:
+            return True
+        return remaining < (_REFRESH_WINDOW_SECONDS - self._refresh_jitter)
 
     async def _acquire_token(self, scope: str) -> _TokenPair:
         """
         Acquire an access token asynchronously for the specified OAuth2 scope.
 
+        Tokens are cached and reused until they are within
+        :data:`~PowerPlatform.Dataverse.core._auth._REFRESH_WINDOW_SECONDS` of
+        expiry.  An :class:`asyncio.Lock` serialises concurrent callers so that
+        only one coroutine calls the underlying credential at a time
+        (double-check locking pattern).
+
         :param scope: OAuth2 scope string, typically ``"https://<org>.crm.dynamics.com/.default"``.
         :type scope: :class:`str`
         :return: Token pair containing the scope and access token.
         :rtype: ~PowerPlatform.Dataverse.core._auth._TokenPair
         :raises ~azure.core.exceptions.ClientAuthenticationError: If token acquisition fails.
         """
-        token = await self.credential.get_token(scope)
-        return _TokenPair(resource=scope, access_token=token.token)
+        if not self._need_new_token():
+            return _TokenPair(resource=scope, access_token=self._token.token)
+        async with self._lock:
+            # Double-check after acquiring lock — another coroutine may have refreshed.
+            if not self._need_new_token():
+                return _TokenPair(resource=scope, access_token=self._token.token)
+            try:
+                self._token = await self.credential.get_token(scope)
+            except Exception as exc:
+                raise AuthenticationError(
+                    f"Token acquisition failed for scope '{scope}'. "
+                    f"Check credential configuration. ({type(exc).__name__})",
+                ) from exc
+            self._refresh_jitter = random.randint(0, _MAX_REFRESH_JITTER_SECONDS)
+            return _TokenPair(resource=scope, access_token=self._token.token)

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

@@ -677,6 +677,11 @@ async def _query_sql(self, sql: str) -> list[dict[str, Any]]:
         # any table resolution.
         sql = self._sql_guardrails(sql)
 
+        # Extract TOP N limit for client-side enforcement.  The Dataverse
+        # ?sql= endpoint ignores TOP in the SQL string, so the SDK must
+        # truncate the result set to honour the caller's intent.
+        top_limit = self._extract_top_limit(sql)
+
         r = await self._execute_raw(await self._build_sql(sql))
         try:
             body = r.json()
@@ -686,14 +691,19 @@ async def _query_sql(self, sql: str) -> list[dict[str, Any]]:
         # Collect first page
         results: list[dict[str, Any]] = []
         if isinstance(body, list):
-            return [row for row in body if isinstance(row, dict)]
+            rows = [row for row in body if isinstance(row, dict)]
+            return rows[:top_limit] if top_limit else rows
         if not isinstance(body, dict):
             return results
 
         value = body.get("value")
         if isinstance(value, list):
             results = [row for row in value if isinstance(row, dict)]
 
+        # Skip pagination entirely when first page already satisfies TOP.
+        if top_limit and len(results) >= top_limit:
+            return results[:top_limit]
+
         # Follow pagination links until exhausted
         raw_link = body.get("@odata.nextLink") or body.get("odata.nextLink")
         next_link: str | None = raw_link if isinstance(raw_link, str) else None
@@ -758,10 +768,13 @@ async def _query_sql(self, sql: str) -> list[dict[str, Any]]:
             if not isinstance(page_value, list) or not page_value:
                 break
             results.extend(row for row in page_value if isinstance(row, dict))
+            # Stop pagination early when TOP limit is reached.
+            if top_limit and len(results) >= top_limit:
+                break
             raw_link = page_body.get("@odata.nextLink") or page_body.get("odata.nextLink")
             next_link = raw_link if isinstance(raw_link, str) else None
 
-        return results
+        return results[:top_limit] if top_limit else results
 
     # ---------------------- Entity set resolution -----------------------
     async def _entity_set_from_schema_name(self, table_schema_name: str) -> str:

src/PowerPlatform/Dataverse/aio/operations/async_records.py

@@ -10,6 +10,7 @@
 from ...core.errors import HttpError
 from ...models.record import QueryResult, Record
 from ...models.upsert import UpsertItem
+from ...operations.records import _validate_guid
 
 if TYPE_CHECKING:
     from ...models.filters import FilterExpression
@@ -93,6 +94,10 @@ async def create(
                 ])
                 print(f"Created {len(guids)} accounts")
         """
+        if not data:
+            raise ValueError(
+                f"data is empty -- provide at least one record with column values to create in '{table}'."
+            )
         async with self._client._scoped_odata() as od:
             entity_set = await od._entity_set_from_schema_name(table)
             if isinstance(data, dict):
@@ -158,10 +163,13 @@ async def update(
             if isinstance(ids, str):
                 if not isinstance(changes, dict):
                     raise TypeError("For single id, changes must be a dict")
+                _validate_guid(ids, "ids")
                 await od._update(table, ids, changes)
                 return None
             if not isinstance(ids, list):
                 raise TypeError("ids must be str or list[str]")
+            for rid in ids:
+                _validate_guid(rid, "ids")
             await od._update_by_ids(table, ids, changes)
             return None
 
@@ -211,6 +219,7 @@ async def delete(
         """
         async with self._client._scoped_odata() as od:
             if isinstance(ids, str):
+                _validate_guid(ids, "ids")
                 await od._delete(table, ids)
                 return None
             if not isinstance(ids, list):
@@ -219,6 +228,8 @@ async def delete(
                 return None
             if not all(isinstance(rid, str) for rid in ids):
                 raise TypeError("ids must contain string GUIDs")
+            for rid in ids:
+                _validate_guid(rid, "ids")
             if use_bulk_delete:
                 return await od._delete_multiple(table, ids)
             for rid in ids:
@@ -269,6 +280,7 @@ async def retrieve(
                 contact = record.get("primarycontactid") or {}
                 print(contact.get("fullname"))
         """
+        _validate_guid(record_id)
         async with self._client._scoped_odata() as od:
             try:
                 raw = await od._get(

src/PowerPlatform/Dataverse/core/_auth.py

@@ -11,9 +11,18 @@
 
 from __future__ import annotations
 
+import random
+import threading
+import time
 from dataclasses import dataclass
+from typing import Optional
 
-from azure.core.credentials import TokenCredential
+from azure.core.credentials import AccessToken, TokenCredential
+
+from .errors import AuthenticationError
+
+_REFRESH_WINDOW_SECONDS = 300  # Proactive refresh 5 min before expiry
+_MAX_REFRESH_JITTER_SECONDS = 60  # Random jitter to spread refresh load
 
 
 @dataclass(repr=False)
@@ -55,16 +64,46 @@ def __init__(self, credential: TokenCredential) -> None:
         if not isinstance(credential, TokenCredential):
             raise TypeError("credential must implement azure.core.credentials.TokenCredential.")
         self.credential: TokenCredential = credential
+        self._token: Optional[AccessToken] = None
+        self._lock = threading.Lock()
+        self._refresh_jitter: int = 0
+
+    def _need_new_token(self) -> bool:
+        """Return ``True`` when the cached token is missing or near expiry."""
+        if self._token is None:
+            return True
+        remaining = self._token.expires_on - time.time()
+        if remaining <= 0:
+            return True
+        return remaining < (_REFRESH_WINDOW_SECONDS - self._refresh_jitter)
 
     def _acquire_token(self, scope: str) -> _TokenPair:
         """
         Acquire an access token for the specified OAuth2 scope.
 
+        Tokens are cached and reused until they are within
+        :data:`_REFRESH_WINDOW_SECONDS` of expiry.  A :class:`threading.Lock`
+        serialises concurrent callers so that only one thread calls the
+        underlying credential at a time (double-check locking pattern).
+
         :param scope: OAuth2 scope string, typically ``"https://<org>.crm.dynamics.com/.default"``.
         :type scope: :class:`str`
         :return: Token pair containing the scope and access token.
         :rtype: ~PowerPlatform.Dataverse.core._auth._TokenPair
         :raises ~azure.core.exceptions.ClientAuthenticationError: If token acquisition fails.
         """
-        token = self.credential.get_token(scope)
-        return _TokenPair(resource=scope, access_token=token.token)
+        if not self._need_new_token():
+            return _TokenPair(resource=scope, access_token=self._token.token)
+        with self._lock:
+            # Double-check after acquiring lock — another thread may have refreshed.
+            if not self._need_new_token():
+                return _TokenPair(resource=scope, access_token=self._token.token)
+            try:
+                self._token = self.credential.get_token(scope)
+            except Exception as exc:
+                raise AuthenticationError(
+                    f"Token acquisition failed for scope '{scope}'. "
+                    f"Check credential configuration. ({type(exc).__name__})",
+                ) from exc
+            self._refresh_jitter = random.randint(0, _MAX_REFRESH_JITTER_SECONDS)
+            return _TokenPair(resource=scope, access_token=self._token.token)

src/PowerPlatform/Dataverse/core/errors.py

@@ -127,6 +127,26 @@ def __init__(self, message: str, *, subcode: Optional[str] = None, details: Opti
         super().__init__(message, code="sql_parse_error", subcode=subcode, details=details, source="client")
 
 
+class AuthenticationError(DataverseError):
+    """
+    Exception raised when token acquisition fails.
+
+    The message is sanitised to prevent credential exceptions from leaking
+    token strings into logs or tracebacks.  The original exception is
+    preserved as ``__cause__`` for advanced debugging.
+
+    :param message: Sanitised human-readable message.
+    :type message: :class:`str`
+    :param subcode: Optional error identifier.
+    :type subcode: :class:`str` | None
+    :param details: Optional additional context.
+    :type details: :class:`dict` | None
+    """
+
+    def __init__(self, message: str, *, subcode: Optional[str] = None, details: Optional[Dict[str, Any]] = None):
+        super().__init__(message, code="authentication_error", subcode=subcode, details=details, source="client")
+
+
 class HttpError(DataverseError):
     """
     Exception raised for HTTP request failures from the Dataverse Web API.
@@ -198,4 +218,4 @@ def __init__(
         )
 
 
-__all__ = ["DataverseError", "HttpError", "ValidationError", "MetadataError", "SQLParseError"]
+__all__ = ["AuthenticationError", "DataverseError", "HttpError", "ValidationError", "MetadataError", "SQLParseError"]

src/PowerPlatform/Dataverse/data/_odata.py

@@ -648,6 +648,11 @@ def _query_sql(self, sql: str) -> list[dict[str, Any]]:
         # any table resolution.
         sql = self._sql_guardrails(sql)
 
+        # Extract TOP N limit for client-side enforcement.  The Dataverse
+        # ?sql= endpoint ignores TOP in the SQL string, so the SDK must
+        # truncate the result set to honour the caller's intent.
+        top_limit = self._extract_top_limit(sql)
+
         r = self._execute_raw(self._build_sql(sql))
         try:
             body = r.json()
@@ -657,14 +662,19 @@ def _query_sql(self, sql: str) -> list[dict[str, Any]]:
         # Collect first page
         results: list[dict[str, Any]] = []
         if isinstance(body, list):
-            return [row for row in body if isinstance(row, dict)]
+            rows = [row for row in body if isinstance(row, dict)]
+            return rows[:top_limit] if top_limit else rows
         if not isinstance(body, dict):
             return results
 
         value = body.get("value")
         if isinstance(value, list):
             results = [row for row in value if isinstance(row, dict)]
 
+        # Skip pagination entirely when first page already satisfies TOP.
+        if top_limit and len(results) >= top_limit:
+            return results[:top_limit]
+
         # Follow pagination links until exhausted
         raw_link = body.get("@odata.nextLink") or body.get("odata.nextLink")
         next_link: str | None = raw_link if isinstance(raw_link, str) else None
@@ -729,10 +739,13 @@ def _query_sql(self, sql: str) -> list[dict[str, Any]]:
             if not isinstance(page_value, list) or not page_value:
                 break
             results.extend(row for row in page_value if isinstance(row, dict))
+            # Stop pagination early when TOP limit is reached.
+            if top_limit and len(results) >= top_limit:
+                break
             raw_link = page_body.get("@odata.nextLink") or page_body.get("odata.nextLink")
             next_link = raw_link if isinstance(raw_link, str) else None
 
-        return results
+        return results[:top_limit] if top_limit else results
 
     # ---------------------- Entity set resolution -----------------------
     def _entity_set_from_schema_name(self, table_schema_name: str) -> str:

src/PowerPlatform/Dataverse/data/_odata_base.py

@@ -801,6 +801,24 @@ def _build_get_relationship(self, schema_name: str) -> _RawRequest:
         r"\bSELECT\b\s+(?:DISTINCT\s+)?(?:TOP\s+\d+(?:\s+PERCENT)?\s+)?\*\s",
         re.IGNORECASE,
     )
+    # Extract TOP N (integer row count) from SELECT.  Does NOT match
+    # TOP N PERCENT -- percentage limits cannot be enforced client-side
+    # because the total row count is unknown before the query completes.
+    _SQL_TOP_RE = re.compile(
+        r"\bSELECT\b\s+(?:DISTINCT\s+)?TOP\s+(\d+)\b(?!\s+PERCENT)",
+        re.IGNORECASE,
+    )
+
+    @staticmethod
+    def _extract_top_limit(sql: str) -> int | None:
+        """Return the integer TOP limit from a SQL SELECT, or ``None``.
+
+        Matches ``SELECT TOP n`` and ``SELECT DISTINCT TOP n`` but
+        **not** ``TOP n PERCENT`` (percentage limits cannot be enforced
+        client-side).
+        """
+        m = _ODataBase._SQL_TOP_RE.search(sql)
+        return int(m.group(1)) if m else None
 
     def _sql_guardrails(self, sql: str) -> str:
         """Apply safety guardrails to a SQL query before sending to the server.

src/PowerPlatform/Dataverse/models/record.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Any, Dict, Iterator, KeysView, List, Optional, Union, ValuesView, ItemsView
+from typing import Any, AsyncIterator, Dict, Iterator, KeysView, List, Optional, Union, ValuesView, ItemsView
 
 __all__ = ["Record", "QueryResult"]
 
@@ -132,6 +132,10 @@ def __init__(self, records: List[Record]) -> None:
     def __iter__(self) -> Iterator[Record]:
         return iter(self.records)
 
+    async def __aiter__(self) -> AsyncIterator[Record]:
+        for record in self.records:
+            yield record
+
     def __len__(self) -> int:
         return len(self.records)