MPT-12358 Collection client async (#19)

Author: albertsola

mpt_api_client/http/collection.py

@@ -1,56 +1,43 @@
 import copy
 from abc import ABC
-from collections.abc import Iterator
+from collections.abc import AsyncIterator, Iterator
 from typing import Any, Self
 
 import httpx
 
-from mpt_api_client.http.client import HTTPClient
+from mpt_api_client.http.client import HTTPClient, HTTPClientAsync
 from mpt_api_client.http.resource import ResourceBaseClient
 from mpt_api_client.models import Collection, Resource
 from mpt_api_client.rql.query_builder import RQLQuery
 
 
-class CollectionBaseClient[ResourceModel: Resource, ResourceClient: ResourceBaseClient[Resource]](  # noqa: WPS214
-    ABC
-):
-    """Immutable Base client for RESTful resource collections.
-
-    Examples:
-        active_orders_cc = order_collection.filter(RQLQuery(status="active"))
-        active_orders = active_orders_cc.order_by("created").iterate()
-        product_active_orders = active_orders_cc.filter(RQLQuery(product__id="PRD-1")).iterate()
-
-        new_order = order_collection.create(order_data)
-
-    """
+class CollectionMixin:
+    """Mixin for collection clients."""
 
     _endpoint: str
-    _resource_class: type[ResourceModel]
-    _resource_client_class: type[ResourceClient]
-    _collection_class: type[Collection[ResourceModel]]
+    _resource_class: type[Any]
+    _resource_client_class: type[Any]
+    _collection_class: type[Collection[Any]]
 
     def __init__(
         self,
+        http_client: HTTPClient | HTTPClientAsync,
         query_rql: RQLQuery | None = None,
-        client: HTTPClient | None = None,
     ) -> None:
-        self.mpt_client = client or HTTPClient()
+        self.http_client = http_client
         self.query_rql: RQLQuery | None = query_rql
         self.query_order_by: list[str] | None = None
         self.query_select: list[str] | None = None
 
     @classmethod
-    def clone(
-        cls, collection_client: "CollectionBaseClient[ResourceModel, ResourceClient]"
-    ) -> Self:
+    def clone(cls, collection_client: "CollectionMixin") -> Self:
         """Create a copy of collection client for immutable operations.
 
         Returns:
             New collection client with same settings.
         """
         new_collection = cls(
-            client=collection_client.mpt_client,
+            http_client=collection_client.http_client,
             query_rql=collection_client.query_rql,
         )
         new_collection.query_order_by = (
@@ -128,6 +115,33 @@ def select(self, *fields: str) -> Self:
         new_client.query_select = list(fields)
         return new_client
 
+
+class CollectionClientBase[ResourceModel: Resource, ResourceClient: ResourceBaseClient[Resource]](  # noqa: WPS214
+    ABC, CollectionMixin
+):
+    """Immutable Base client for RESTful resource collections.
+
+    Examples:
+        active_orders_cc = order_collection.filter(RQLQuery(status="active"))
+        active_orders = active_orders_cc.order_by("created").iterate()
+        product_active_orders = active_orders_cc.filter(RQLQuery(product__id="PRD-1")).iterate()
+
+        new_order = order_collection.create(order_data)
+
+    """
+
+    _resource_class: type[ResourceModel]
+    _resource_client_class: type[ResourceClient]
+    _collection_class: type[Collection[ResourceModel]]
+
+    def __init__(
+        self,
+        query_rql: RQLQuery | None = None,
+        http_client: HTTPClient | None = None,
+    ) -> None:
+        self.http_client: HTTPClient = http_client or HTTPClient()  # type: ignore[mutable-override]
+        CollectionMixin.__init__(self, http_client=self.http_client, query_rql=query_rql)
+
     def fetch_page(self, limit: int = 100, offset: int = 0) -> Collection[ResourceModel]:
         """Fetch one page of resources.
 
@@ -185,15 +199,15 @@ def iterate(self, batch_size: int = 100) -> Iterator[ResourceModel]:
 
     def get(self, resource_id: str) -> ResourceClient:
         """Get resource by resource_id."""
-        return self._resource_client_class(client=self.mpt_client, resource_id=resource_id)
+        return self._resource_client_class(http_client=self.http_client, resource_id=resource_id)
 
     def create(self, resource_data: dict[str, Any]) -> ResourceModel:
         """Create a new resource using `POST /endpoint`.
 
         Returns:
             New resource created.
         """
-        response = self.mpt_client.post(self._endpoint, json=resource_data)
+        response = self.http_client.post(self._endpoint, json=resource_data)
         response.raise_for_status()
 
         return self._resource_class.from_response(response)
@@ -208,7 +222,121 @@ def _fetch_page_as_response(self, limit: int = 100, offset: int = 0) -> httpx.Re
             HTTPStatusError: if the response status code is not 200.
         """
         pagination_params: dict[str, int] = {"limit": limit, "offset": offset}
-        response = self.mpt_client.get(self.build_url(pagination_params))
+        response = self.http_client.get(self.build_url(pagination_params))
+        response.raise_for_status()
+
+        return response
+
+
+class AsyncCollectionClientBase[
+    ResourceModel: Resource,
+    ResourceClient: ResourceBaseClient[Resource],
+](ABC, CollectionMixin):
+    """Immutable Base client for RESTful resource collections.
+
+    Examples:
+        active_orders_cc = order_collection.filter(RQLQuery(status="active"))
+        active_orders = active_orders_cc.order_by("created").iterate()
+        product_active_orders = active_orders_cc.filter(RQLQuery(product__id="PRD-1")).iterate()
+
+        new_order = order_collection.create(order_data)
+
+    """
+
+    _resource_class: type[ResourceModel]
+    _resource_client_class: type[ResourceClient]
+    _collection_class: type[Collection[ResourceModel]]
+
+    def __init__(
+        self,
+        query_rql: RQLQuery | None = None,
+        http_client: HTTPClientAsync | None = None,
+    ) -> None:
+        self.http_client: HTTPClientAsync = http_client or HTTPClientAsync()  # type: ignore[mutable-override]
+        CollectionMixin.__init__(self, http_client=self.http_client, query_rql=query_rql)
+
+    async def fetch_page(self, limit: int = 100, offset: int = 0) -> Collection[ResourceModel]:
+        """Fetch one page of resources.
+
+        Returns:
+            Collection of resources.
+        """
+        response = await self._fetch_page_as_response(limit=limit, offset=offset)
+        return Collection.from_response(response)
+
+    async def fetch_one(self) -> ResourceModel:
+        """Fetch one page, expect exactly one result.
+
+        Returns:
+            One resource.
+
+        Raises:
+            ValueError: If the total matching records are not exactly one.
+        """
+        response = await self._fetch_page_as_response(limit=1, offset=0)
+        resource_list: Collection[ResourceModel] = Collection.from_response(response)
+        total_records = len(resource_list)
+        if resource_list.meta:
+            total_records = resource_list.meta.pagination.total
+        if total_records == 0:
+            raise ValueError("Expected one result, but got zero results")
+        if total_records > 1:
+            raise ValueError(f"Expected one result, but got {total_records} results")
+
+        return resource_list[0]
+
+    async def iterate(self, batch_size: int = 100) -> AsyncIterator[ResourceModel]:
+        """Iterate over all resources, yielding GenericResource objects.
+
+        Args:
+            batch_size: Number of resources to fetch per request
+
+        Returns:
+            Iterator of resources.
+        """
+        offset = 0
+        limit = batch_size  # Default page size
+
+        while True:
+            response = await self._fetch_page_as_response(limit=limit, offset=offset)
+            items_collection: Collection[ResourceModel] = self._collection_class.from_response(
+                response
+            )
+            for resource in items_collection:
+                yield resource
+
+            if not items_collection.meta:
+                break
+            if not items_collection.meta.pagination.has_next():
+                break
+            offset = items_collection.meta.pagination.next_offset()
+
+    async def get(self, resource_id: str) -> ResourceClient:
+        """Get resource by resource_id."""
+        return self._resource_client_class(http_client=self.http_client, resource_id=resource_id)  # type: ignore[arg-type]
+
+    async def create(self, resource_data: dict[str, Any]) -> ResourceModel:
+        """Create a new resource using `POST /endpoint`.
+
+        Returns:
+            New resource created.
+        """
+        response = await self.http_client.post(self._endpoint, json=resource_data)
+        response.raise_for_status()
+
+        return self._resource_class.from_response(response)
+
+    async def _fetch_page_as_response(self, limit: int = 100, offset: int = 0) -> httpx.Response:
+        """Fetch one page of resources.
+
+        Returns:
+            httpx.Response object.
+
+        Raises:
+            HTTPStatusError: if the response status code is not 200.
+        """
+        pagination_params: dict[str, int] = {"limit": limit, "offset": offset}
+        response = await self.http_client.get(self.build_url(pagination_params))
         response.raise_for_status()
 
         return response

mpt_api_client/http/resource.py

@@ -12,10 +12,10 @@ class ResourceBaseClient[ResourceModel: Resource](ABC):  # noqa: WPS214
 
     _endpoint: str
     _resource_class: type[ResourceModel]
-    _safe_attributes: ClassVar[set[str]] = {"mpt_client_", "resource_id_", "resource_"}
+    _safe_attributes: ClassVar[set[str]] = {"http_client_", "resource_id_", "resource_"}
 
-    def __init__(self, client: HTTPClient, resource_id: str) -> None:
-        self.mpt_client_ = client  # noqa: WPS120
+    def __init__(self, http_client: HTTPClient, resource_id: str) -> None:
+        self.http_client_ = http_client  # noqa: WPS120
         self.resource_id_ = resource_id  # noqa: WPS120
         self.resource_: Resource | None = None  # noqa: WPS120
 
@@ -78,7 +78,7 @@ def do_action(
             HTTPError: If the action fails.
         """
         url = f"{self.resource_url}/{url}" if url else self.resource_url
-        response = self.mpt_client_.request(method, url, json=json)
+        response = self.http_client_.request(method, url, json=json)
         response.raise_for_status()
         return response
 

mpt_api_client/mptclient.py

@@ -1,6 +1,6 @@
 from mpt_api_client.http.client import HTTPClient
 from mpt_api_client.registry import Registry, commerce
-from mpt_api_client.resources import OrderCollectionClient
+from mpt_api_client.resources import OrderCollectionClientBase
 
 
 class MPTClientBase:
@@ -11,13 +11,13 @@ def __init__(
         base_url: str | None = None,
         api_key: str | None = None,
         registry: Registry | None = None,
-        mpt_client: HTTPClient | None = None,
+        http_client: HTTPClient | None = None,
     ):
-        self.mpt_client = mpt_client or HTTPClient(base_url=base_url, api_token=api_key)
+        self.http_client = http_client or HTTPClient(base_url=base_url, api_token=api_key)
         self.registry: Registry = registry or Registry()
 
     def __getattr__(self, name):  # type: ignore[no-untyped-def]
-        return self.registry.get(name)(client=self.mpt_client)
+        return self.registry.get(name)(http_client=self.http_client)
 
 
 class MPTClient(MPTClientBase):
@@ -31,14 +31,14 @@ def commerce(self) -> "CommerceMpt":
         for managing agreements, requests, subscriptions, and orders
         within a vendor-client-ops ecosystem.
         """
-        return CommerceMpt(mpt_client=self.mpt_client, registry=commerce)
+        return CommerceMpt(http_client=self.http_client, registry=commerce)
 
 
 class CommerceMpt(MPTClientBase):
     """Commerce MPT API Client."""
 
     @property
-    def orders(self) -> OrderCollectionClient:
+    def orders(self) -> OrderCollectionClientBase:
         """Orders MPT API collection.
 
         The Orders API provides a comprehensive set of endpoints
@@ -54,4 +54,4 @@ def orders(self) -> OrderCollectionClient:
                [...]
 
         """
-        return self.registry.get("orders")(client=self.mpt_client)  # type: ignore[return-value]
+        return self.registry.get("orders")(http_client=self.http_client)  # type: ignore[return-value]

mpt_api_client/registry.py

@@ -1,9 +1,9 @@
 from collections.abc import Callable
 from typing import Any
 
-from mpt_api_client.http.collection import CollectionBaseClient
+from mpt_api_client.http.collection import CollectionClientBase
 
-ItemType = type[CollectionBaseClient[Any, Any]]
+ItemType = type[CollectionClientBase[Any, Any]]
 
 
 class Registry:

mpt_api_client/resources/__init__.py

@@ -1,3 +1,3 @@
-from mpt_api_client.resources.order import Order, OrderCollectionClient, OrderResourceClient
+from mpt_api_client.resources.order import Order, OrderCollectionClientBase, OrderResourceClient
 
-__all__ = ["Order", "OrderCollectionClient", "OrderResourceClient"]  # noqa: WPS410
+__all__ = ["Order", "OrderCollectionClientBase", "OrderResourceClient"]  # noqa: WPS410

mpt_api_client/resources/order.py

@@ -1,6 +1,6 @@
 from typing import Any
 
-from mpt_api_client.http.collection import CollectionBaseClient
+from mpt_api_client.http.collection import CollectionClientBase
 from mpt_api_client.http.resource import ResourceBaseClient
 from mpt_api_client.models import Collection, Resource
 from mpt_api_client.registry import commerce
@@ -76,7 +76,7 @@ def template(self) -> str:
 
 
 @commerce("orders")
-class OrderCollectionClient(CollectionBaseClient[Order, OrderResourceClient]):
+class OrderCollectionClientBase(CollectionClientBase[Order, OrderResourceClient]):
     """Orders client."""
 
     _endpoint = "/public/v1/commerce/orders"