MPT-12327 Implement collection client

Author: albertsola

mpt_api_client/http/client.py

@@ -1,3 +1,5 @@
+import os
+
 import httpx
 
 
@@ -7,12 +9,26 @@ class MPTClient(httpx.Client):
     def __init__(
         self,
         *,
-        base_url: str,
-        api_token: str,
+        base_url: str | None = None,
+        api_token: str | None = None,
         timeout: float = 5.0,
         retries: int = 0,
     ):
-        self.api_token = api_token
+        api_token = api_token or os.getenv("MPT_TOKEN")
+        if not api_token:
+            raise ValueError(
+                "API token is required. "
+                "Set it up as env variable MPT_TOKEN or pass it as `api_token` "
+                "argument to MPTClient."
+            )
+
+        base_url = base_url or os.getenv("MPT_URL")
+        if not base_url:
+            raise ValueError(
+                "Base URL is required. "
+                "Set it up as env variable MPT_URL or pass it as `base_url` "
+                "argument to MPTClient."
+            )
         base_headers = {
             "User-Agent": "swo-marketplace-client/1.0",
             "Authorization": f"Bearer {api_token}",

mpt_api_client/http/collection.py

@@ -0,0 +1,199 @@
+import copy
+from abc import ABC
+from collections.abc import Iterator
+from typing import Any, Self
+
+import httpx
+
+from mpt_api_client.http.client import MPTClient
+from mpt_api_client.models import Collection, Resource
+from mpt_api_client.rql.query_builder import RQLQuery
+
+
+class CollectionBaseClient[ResourceType: Resource](ABC):  # noqa: WPS214
+    """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)
+
+    """
+
+    _endpoint: str
+    _resource_class: type[ResourceType]
+    _collection_class: type[Collection[ResourceType]]
+
+    def __init__(
+        self,
+        query_rql: RQLQuery | None = None,
+        client: MPTClient | None = None,
+    ) -> None:
+        self.mpt_client = client or MPTClient()
+        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[ResourceType]") -> 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,
+            query_rql=collection_client.query_rql,
+        )
+        new_collection.query_order_by = (
+            copy.copy(collection_client.query_order_by)
+            if collection_client.query_order_by
+            else None
+        )
+        new_collection.query_select = (
+            copy.copy(collection_client.query_select) if collection_client.query_select else None
+        )
+        return new_collection
+
+    def build_url(self, query_params: dict[str, Any] | None = None) -> str:
+        """Builds the endpoint URL with all the query parameters.
+
+        Returns:
+            Partial URL with query parameters.
+        """
+        query_params = query_params or {}
+        query_parts = [
+            f"{param_key}={param_value}" for param_key, param_value in query_params.items()
+        ]  # noqa: WPS237
+        if self.query_order_by:
+            query_parts.append(f"order={','.join(self.query_order_by)}")  # noqa: WPS237
+        if self.query_select:
+            query_parts.append(f"select={','.join(self.query_select)}")  # noqa: WPS237
+        if self.query_rql:
+            query_parts.append(str(self.query_rql))
+        if query_parts:
+            return f"{self._endpoint}?{'&'.join(query_parts)}"  # noqa: WPS237
+        return self._endpoint
+
+    def order_by(self, *fields: str) -> Self:
+        """Returns new collection with ordering setup.
+
+        Returns:
+            New collection with ordering setup.
+
+        Raises:
+            ValueError: If ordering has already been set.
+        """
+        if self.query_order_by is not None:
+            raise ValueError("Ordering is already set. Cannot set ordering multiple times.")
+        new_collection = self.clone(self)
+        new_collection.query_order_by = list(fields)
+        return new_collection
+
+    def filter(self, rql: RQLQuery) -> Self:
+        """Creates a new collection with the filter added to the filter collection.
+
+        Returns:
+            New copy of the collection with the filter added.
+        """
+        if self.query_rql:
+            rql = self.query_rql & rql
+        new_collection = self.clone(self)
+        new_collection.query_rql = rql
+        return new_collection
+
+    def select(self, *fields: str) -> Self:
+        """Set select fields. Raises ValueError if select fields are already set.
+
+        Returns:
+            New copy of the collection with the select fields set.
+
+        Raises:
+            ValueError: If select fields are already set.
+        """
+        if self.query_select is not None:
+            raise ValueError(
+                "Select fields are already set. Cannot set select fields multiple times."
+            )
+
+        new_client = self.clone(self)
+        new_client.query_select = list(fields)
+        return new_client
+
+    def fetch_page(self, limit: int = 100, offset: int = 0) -> Collection[ResourceType]:
+        """Fetch one page of resources.
+
+        Returns:
+            Collection of resources.
+        """
+        response = self._fetch_page_as_response(limit=limit, offset=offset)
+        return Collection.from_response(response)
+
+    def fetch_one(self) -> ResourceType:
+        """Fetch one page, expect exactly one result.
+
+        Returns:
+            One resource.
+
+        Raises:
+            ValueError: If the total matching records are not exactly one.
+        """
+        response = self._fetch_page_as_response(limit=1, offset=0)
+        resource_list: Collection[ResourceType] = 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]
+
+    def iterate(self) -> Iterator[ResourceType]:
+        """Iterate over all resources, yielding GenericResource objects.
+
+        Returns:
+            Iterator of resources.
+        """
+        offset = 0
+        limit = 100  # Default page size
+
+        while True:
+            response = self._fetch_page_as_response(limit=limit, offset=offset)
+            items_collection: Collection[ResourceType] = Collection.from_response(response)
+            yield from items_collection
+
+            if not items_collection.meta:
+                break
+            if not items_collection.meta.pagination.has_next():
+                break
+            offset = items_collection.meta.pagination.next_offset()
+
+    def create(self, resource_data: dict[str, Any]) -> ResourceType:
+        """Create a new resource using `POST /endpoint`.
+
+        Returns:
+            New resource created.
+        """
+        response = self.mpt_client.post(self._endpoint, json=resource_data)
+        response.raise_for_status()
+
+        return self._resource_class.from_response(response)
+
+    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 = self.mpt_client.get(self.build_url(pagination_params))
+        response.raise_for_status()
+
+        return response

mpt_api_client/modules/__init__.py

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

mpt_api_client/modules/order.py

@@ -0,0 +1,16 @@
+from mpt_api_client.http.collection import CollectionBaseClient
+from mpt_api_client.models import Collection, Resource
+from mpt_api_client.register import mpt
+
+
+class Order(Resource):
+    """Order resource."""
+
+
+@mpt("orders")
+class OrderCollectionClient(CollectionBaseClient[Order]):
+    """Orders client."""
+
+    _endpoint = "/api/v1/orders"
+    _resource_class = Order
+    _collection_class = Collection[Order]

mpt_api_client/mpt.py

@@ -0,0 +1,8 @@
+from mpt_api_client.http.client import MPTClient
+
+
+class MPT:
+    """MPT API Client."""
+
+    def __init__(self, base_url: str | None = None, api_key: str | None = None):
+        self.mpt_client = MPTClient(base_url=base_url, api_token=api_key)

mpt_api_client/register.py

@@ -0,0 +1,74 @@
+from collections.abc import Callable
+from typing import Any
+
+from mpt_api_client.http.collection import CollectionBaseClient
+
+ItemType = type[CollectionBaseClient[Any]]
+
+
+class Registry:
+    """Registry for MPT collection clients."""
+
+    def __init__(self) -> None:
+        self.items: dict[str, ItemType] = {}  # noqa: WPS110
+
+    def __call__(self, keyname: str) -> Callable[[ItemType], ItemType]:
+        """Decorator to register a CollectionBaseClient class.
+
+        Args:
+            keyname: The key to register the class under
+
+        Returns:
+            The decorator function
+
+        Examples:
+            registry = Registry()
+            @registry("orders")
+            class OrderCollectionClient(CollectionBaseClient):
+                _endpoint = "/api/v1/orders"
+                _resource_class = Order
+
+            registry.get("orders") == OrderCollectionClient
+        """
+
+        def decorator(cls: ItemType) -> ItemType:
+            self.register(keyname, cls)
+            return cls
+
+        return decorator
+
+    def register(self, keyname: str, item: ItemType) -> None:  # noqa: WPS110
+        """Register a collection client class with a keyname.
+
+        Args:
+            keyname: The key to register the client under
+            item: The collection client class to register
+        """
+        self.items[keyname] = item
+
+    def get(self, keyname: str) -> ItemType:
+        """Get a registered collection client class by keyname.
+
+        Args:
+            keyname: The key to look up
+
+        Returns:
+            The registered collection client class
+
+        Raises:
+            KeyError: If keyname is not registered
+        """
+        if keyname not in self.items:
+            raise KeyError(f"No collection client registered with keyname: {keyname}")
+        return self.items[keyname]
+
+    def list_keys(self) -> list[str]:
+        """Get all registered keynames.
+
+        Returns:
+            List of all registered keynames
+        """
+        return list(self.items.keys())
+
+
+mpt = Registry()

setup.cfg

@@ -37,6 +37,14 @@ per-file-ignores =
     WPS110
     # Found `noqa` comments overuse
     WPS402
+  tests/http/collection/test_collection_client_iterate.py:
+    # Found too many module members
+    WPS202
+  tests/http/collection/test_collection_client_fetch.py:
+    # Found too many module members
+    WPS202
+    # Found magic number
+    WPS432
   tests/*:
     # Allow magic strings
     WPS432

tests/http/collection/conftest.py

@@ -0,0 +1,19 @@
+import pytest
+
+from mpt_api_client.http.collection import CollectionBaseClient
+from mpt_api_client.models import Collection, Resource
+
+
+class DummyResource(Resource):
+    """Dummy resource for testing."""
+
+
+class DummyCollectionClient(CollectionBaseClient[DummyResource]):
+    _endpoint = "/api/v1/test"
+    _resource_class = DummyResource
+    _collection_class = Collection[DummyResource]
+
+
+@pytest.fixture
+def collection_client(mpt_client):
+    return DummyCollectionClient(client=mpt_client)