Add async/await support to Dataverse SDK via inheritance-based AsyncDataverseClient

Co-authored-by: saurabhrb <32964911+saurabhrb@users.noreply.github.com>

Author: Copilot

architecture/async_design.md

@@ -0,0 +1,177 @@
+# Async Architecture Design
+
+## Design Philosophy
+
+The async layer follows the **inheritance-based async pattern** used by the Azure SDK for Python:
+a parallel class hierarchy that inherits all pure-logic from the sync classes and overrides only
+the three blocking operations.
+
+### Only 3 blocking operations get async overrides
+
+| Blocking operation | Sync implementation | Async implementation |
+|---|---|---|
+| Token acquisition | `credential.get_token(scope)` | `await credential.get_token(scope)` |
+| HTTP I/O | `requests.request(...)` | `await aiohttp.ClientSession.request(...)` |
+| Sleep / backoff | `time.sleep(delay)` | `await asyncio.sleep(delay)` |
+
+Everything else — URL building, OData serialization, key formatting, cache lookups,
+payload construction, error parsing — is **pure CPU logic** that runs in microseconds.
+These methods are inherited directly from the sync classes with no override needed.
+
+## Class Hierarchy
+
+```
+azure.core.credentials.TokenCredential
+    _AuthManager._acquire_token(scope)  →  credential.get_token(scope)
+
+azure.core.credentials_async.AsyncTokenCredential
+    _AsyncAuthManager._acquire_token(scope)  →  await credential.get_token(scope)
+
+_HttpClient._request(method, url, **kw)  →  requests.request(...)
+    _ODataClient._raw_request(...)
+        _ODataClient._request(...)
+            ... all sync CRUD / metadata methods
+
+_AsyncHttpClient._request(method, url, **kw)  →  await aiohttp.ClientSession.request(...)
+    _AsyncODataClient(inherits _ODataClient)
+        override: _raw_request, _request, _headers, _merge_headers
+        override: _create, _create_multiple, _update, _update_multiple, ...
+        override: _entity_set_from_schema_name, _get, _get_multiple, ...
+        inherited: _format_key, _build_alternate_key_str, _escape_odata_quotes, ...
+        inherited: _attribute_payload, _label, _to_pascal, _normalize_cache_key, ...
+
+DataverseClient
+    records → RecordOperations (sync)
+    query   → QueryOperations (sync)
+    tables  → TableOperations (sync)
+    files   → FileOperations (sync)
+
+AsyncDataverseClient
+    records → AsyncRecordOperations (async)
+    query   → AsyncQueryOperations (async)
+    tables  → AsyncTableOperations (async)
+    files   → AsyncFileOperations (async)
+```
+
+## File Map
+
+| File | Purpose |
+|---|---|
+| `core/_auth.py` | Sync `_AuthManager` (unchanged) |
+| `core/_async_auth.py` | Async `_AsyncAuthManager` |
+| `core/_http.py` | Sync `_HttpClient` using `requests` (unchanged) |
+| `core/_async_http.py` | Async `_AsyncHttpClient` using `aiohttp` |
+| `data/_odata.py` | Sync `_ODataClient` (unchanged) |
+| `data/_async_odata.py` | Async `_AsyncODataClient` inheriting `_ODataClient` |
+| `client.py` | Sync `DataverseClient` (unchanged) |
+| `async_client.py` | Async `AsyncDataverseClient` |
+| `operations/records.py` | Sync `RecordOperations` (unchanged) |
+| `operations/async_records.py` | Async `AsyncRecordOperations` |
+| `operations/query.py` | Sync `QueryOperations` (unchanged) |
+| `operations/async_query.py` | Async `AsyncQueryOperations` |
+| `operations/tables.py` | Sync `TableOperations` (unchanged) |
+| `operations/async_tables.py` | Async `AsyncTableOperations` |
+| `operations/files.py` | Sync `FileOperations` (unchanged) |
+| `operations/async_files.py` | Async `AsyncFileOperations` |
+
+## Async HTTP Response Wrapper
+
+`_AsyncODataClient` depends on response objects that provide `.status_code`, `.headers`,
+`.text`, and `.json()` synchronously (matching the `requests.Response` interface used
+throughout `_ODataClient`).
+
+`_AsyncResponse` achieves this by **eagerly reading the entire response body** when the
+aiohttp request completes. This is acceptable for Dataverse API responses (which are
+typically small JSON payloads). File uploads use streaming writes (not reads), so
+eager body reading does not affect upload performance.
+
+## Async Generator for `_get_multiple`
+
+The sync `_get_multiple` is a regular generator (`yield`). The async version is an
+**async generator** (`async def` with `yield`), enabling callers to iterate pages with
+`async for`:
+
+```python
+async for page in od._get_multiple("account", filter="statecode eq 0"):
+    for row in page:
+        print(row["name"])
+```
+
+At the public API level, `AsyncRecordOperations.get()` returns an async generator function:
+
+```python
+pages = await client.records.get("account", filter="statecode eq 0")
+async for page in pages:
+    for record in page:
+        print(record["name"])
+```
+
+## ContextVar Correlation IDs
+
+`_CALL_SCOPE_CORRELATION_ID` is a `ContextVar`. Python's `ContextVar` is fully compatible
+with asyncio — each task gets its own copy of the context. The `_call_scope()` context
+manager (sync `@contextmanager`) is used inside `@asynccontextmanager` (`_scoped_odata`)
+via a regular `with` statement — this is safe because the ContextVar set/reset is
+instantaneous (no I/O).
+
+## Usage Comparison
+
+### Sync (existing code — unchanged)
+
+```python
+from azure.identity import ClientSecretCredential
+from PowerPlatform.Dataverse.client import DataverseClient
+
+credential = ClientSecretCredential(tenant_id, client_id, client_secret)
+
+with DataverseClient("https://org.crm.dynamics.com", credential) as client:
+    guid = client.records.create("account", {"name": "Contoso"})
+    record = client.records.get("account", guid)
+    client.records.update("account", guid, {"telephone1": "555-0100"})
+    client.records.delete("account", guid)
+```
+
+### Async (new)
+
+```python
+import asyncio
+from azure.identity.aio import ClientSecretCredential
+from PowerPlatform.Dataverse.async_client import AsyncDataverseClient
+
+credential = ClientSecretCredential(tenant_id, client_id, client_secret)
+
+async def main():
+    async with AsyncDataverseClient("https://org.crm.dynamics.com", credential) as client:
+        guid = await client.records.create("account", {"name": "Contoso"})
+        record = await client.records.get("account", guid)
+        await client.records.update("account", guid, {"telephone1": "555-0100"})
+        await client.records.delete("account", guid)
+
+asyncio.run(main())
+```
+
+## Installation
+
+### Async support (optional dependency)
+
+```bash
+pip install "PowerPlatform-Dataverse-Client[async]"
+```
+
+This installs `aiohttp>=3.13.3` in addition to the core dependencies. The sync client
+continues to work without `aiohttp` installed.
+
+## Migration Guide
+
+Existing sync code does **not** need to change. Async support is purely additive:
+
+1. Change import: `from PowerPlatform.Dataverse.client import DataverseClient`
+   → `from PowerPlatform.Dataverse.async_client import AsyncDataverseClient`
+
+2. Use `async with` instead of `with`
+
+3. Use async credentials: `azure.identity.aio.*` instead of `azure.identity.*`
+
+4. Add `await` before every operation call
+
+5. Use `async for` when iterating pages returned by `client.records.get(table, ...)`

examples/basic/async_example.py

@@ -0,0 +1,50 @@
+"""Basic async usage example for the Dataverse SDK."""
+
+import asyncio
+
+# Use an async Azure Identity credential
+# Install: pip install azure-identity
+from azure.identity.aio import InteractiveBrowserCredential
+
+from PowerPlatform.Dataverse.async_client import AsyncDataverseClient
+
+
+async def main() -> None:
+    credential = InteractiveBrowserCredential()
+
+    async with AsyncDataverseClient("https://org.crm.dynamics.com", credential) as client:
+        # Create a single record
+        guid = await client.records.create("account", {"name": "Contoso"})
+        print(f"[OK] Created account: {guid}")
+
+        # Fetch the record back
+        record = await client.records.get("account", guid, select=["name"])
+        print(f"[OK] Name: {record['name']}")
+
+        # Update the record
+        await client.records.update("account", guid, {"telephone1": "555-0100"})
+        print("[OK] Updated telephone")
+
+        # Delete the record
+        await client.records.delete("account", guid)
+        print("[OK] Deleted account")
+
+        # SQL query (async)
+        rows = await client.query.sql("SELECT TOP 5 name FROM account ORDER BY name")
+        for row in rows:
+            print(f"  account: {row['name']}")
+
+        # Multi-record fetch with async pagination
+        print("[INFO] Paging through active accounts:")
+        async for page in await client.records.get(
+            "account",
+            filter="statecode eq 0",
+            select=["name"],
+            page_size=20,
+        ):
+            for rec in page:
+                print(f"  {rec['name']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -41,13 +41,16 @@ dependencies = [
 dataverse-install-claude-skill = "PowerPlatform.Dataverse._skill_installer:main"
 
 [project.optional-dependencies]
+async = ["aiohttp>=3.13.3"]
 dev = [
     "pytest>=7.0.0",
     "pytest-cov>=4.0.0",
+    "pytest-asyncio>=0.23.0",
     "black>=23.0.0",
     "isort>=5.12.0",
     "mypy>=1.0.0",
     "ruff>=0.1.0",
+    "aiohttp>=3.13.3",
 ]
 
 [tool.setuptools]

src/PowerPlatform/Dataverse/async_client.py

@@ -0,0 +1,142 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""
+Async Dataverse client.
+
+:class:`~PowerPlatform.Dataverse.async_client.AsyncDataverseClient` mirrors the public API of
+:class:`~PowerPlatform.Dataverse.client.DataverseClient` with full ``async``/``await`` support.
+Existing sync code is completely unaffected; async support is opt-in via a separate import.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any, AsyncIterator, Dict, List, Optional, Union
+
+from azure.core.credentials_async import AsyncTokenCredential
+
+from .core._async_auth import _AsyncAuthManager
+from .core.config import DataverseConfig
+from .data._async_odata import _AsyncODataClient
+from .operations.async_records import AsyncRecordOperations
+from .operations.async_query import AsyncQueryOperations
+from .operations.async_tables import AsyncTableOperations
+from .operations.async_files import AsyncFileOperations
+
+__all__ = ["AsyncDataverseClient"]
+
+
+class AsyncDataverseClient:
+    """
+    Async high-level client for Microsoft Dataverse operations.
+
+    Mirrors :class:`~PowerPlatform.Dataverse.client.DataverseClient` with ``async``/``await``
+    support. All methods are ``async def`` and must be awaited.
+
+    :param base_url: Your Dataverse environment URL, for example
+        ``"https://org.crm.dynamics.com"``. Trailing slash is automatically removed.
+    :type base_url: :class:`str`
+    :param credential: Azure Identity async credential for authentication.
+    :type credential: ~azure.core.credentials_async.AsyncTokenCredential
+    :param config: Optional configuration for language, timeouts, and retries.
+        If not provided, defaults are loaded from
+        :meth:`~PowerPlatform.Dataverse.core.config.DataverseConfig.from_env`.
+    :type config: ~PowerPlatform.Dataverse.core.config.DataverseConfig or None
+
+    :raises ValueError: If ``base_url`` is missing or empty after trimming.
+
+    Operations are organized into namespaces:
+
+    - ``client.records`` -- create, update, delete, and get records (single or paginated)
+    - ``client.query``   -- query and search operations
+    - ``client.tables``  -- table and column metadata management
+    - ``client.files``   -- file upload operations
+
+    The client supports Python's async context manager protocol::
+
+        from azure.identity.aio import ClientSecretCredential
+        from PowerPlatform.Dataverse.async_client import AsyncDataverseClient
+
+        credential = ClientSecretCredential(tenant_id, client_id, client_secret)
+
+        async with AsyncDataverseClient("https://org.crm.dynamics.com", credential) as client:
+            guid = await client.records.create("account", {"name": "Contoso Ltd"})
+            record = await client.records.get("account", guid, select=["name"])
+            print(record["name"])
+            await client.records.delete("account", guid)
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        credential: AsyncTokenCredential,
+        config: Optional[DataverseConfig] = None,
+    ) -> None:
+        self.auth = _AsyncAuthManager(credential)
+        self._base_url = (base_url or "").rstrip("/")
+        if not self._base_url:
+            raise ValueError("base_url is required.")
+        self._config = config or DataverseConfig.from_env()
+        self._odata: Optional[_AsyncODataClient] = None
+        self._closed: bool = False
+
+        # Operation namespaces
+        self.records = AsyncRecordOperations(self)
+        self.query = AsyncQueryOperations(self)
+        self.tables = AsyncTableOperations(self)
+        self.files = AsyncFileOperations(self)
+
+    def _get_odata(self) -> _AsyncODataClient:
+        """Get or lazily create the internal async OData client."""
+        if self._odata is None:
+            self._odata = _AsyncODataClient(
+                self.auth,
+                self._base_url,
+                self._config,
+            )
+        return self._odata
+
+    @asynccontextmanager
+    async def _scoped_odata(self) -> AsyncIterator[_AsyncODataClient]:
+        """Yield the async OData client with an active correlation scope."""
+        self._check_closed()
+        od = self._get_odata()
+        with od._call_scope():
+            yield od
+
+    # ---------------- Context manager / lifecycle ----------------
+
+    async def __aenter__(self) -> AsyncDataverseClient:
+        """Enter the async context manager.
+
+        :return: The client instance.
+        :rtype: AsyncDataverseClient
+        :raises RuntimeError: If the client has been closed.
+        """
+        self._check_closed()
+        return self
+
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        """Exit the async context manager and close the client."""
+        await self.close()
+
+    async def close(self) -> None:
+        """Close the client and release resources.
+
+        Closes the underlying aiohttp session, clears caches, and marks the
+        client as closed. Safe to call multiple times.
+
+        Called automatically when using the client as an async context manager.
+        """
+        if self._closed:
+            return
+        if self._odata is not None:
+            await self._odata.close()
+            self._odata = None
+        self._closed = True
+
+    def _check_closed(self) -> None:
+        """Raise :class:`RuntimeError` if the client has been closed."""
+        if self._closed:
+            raise RuntimeError("AsyncDataverseClient is closed")