MPT-12327 Implement collection model

Author: albertsola

mpt_api_client/http/models.py

@@ -0,0 +1,4 @@
+from mpt_api_client.models.collection import Collection
+from mpt_api_client.models.resource import Resource
+
+__all__ = ["Collection", "Resource"]  # noqa: WPS410

mpt_api_client/models/__init__.py

@@ -0,0 +1,54 @@
+from collections.abc import Iterator
+from typing import Any, ClassVar, Self
+
+from httpx import Response
+
+from mpt_api_client.models.meta import Meta
+from mpt_api_client.models.resource import Resource
+from mpt_api_client.models.types import BaseCollection, BaseResource, ResourceData
+
+
+class Collection[ResourceModel](BaseCollection):
+    """Provides a base collection to interact with api collection data using fluent interfaces."""
+
+    _data_key: ClassVar[str] = "data"
+    _resource_model: ClassVar[type[BaseResource]] = Resource
+
+    def __init__(
+        self, collection_data: list[ResourceData] | None = None, meta: Meta | None = None
+    ) -> None:
+        self.meta = meta
+        collection_data = collection_data or []
+        self._resource_collection = [
+            self._resource_model(resource_data, meta) for resource_data in collection_data
+        ]
+
+    def __getitem__(self, index: int) -> ResourceModel:
+        """Returns the collection item at the given index."""
+        return self._resource_collection[index]
+
+    def __iter__(self) -> Iterator[ResourceModel]:
+        """Make GenericCollection iterable."""
+        return iter(self._resource_collection)
+
+    def __len__(self) -> int:
+        """Return the number of items in the collection."""
+        return len(self._resource_collection)
+
+    def __bool__(self) -> bool:
+        """Returns True if collection has items."""
+        return len(self._resource_collection) > 0
+
+    @classmethod
+    def from_response(cls, response: Response) -> Self:
+        """Creates a collection from a response."""
+        response_data = response.json().get(cls._data_key)
+        meta = Meta.from_response(response)
+        if not isinstance(response_data, list):
+            raise TypeError(f"Response `{cls._data_key}` must be a list for collection endpoints.")
+
+        return cls(response_data, meta)
+
+    def to_list(self) -> list[dict[str, Any]]:
+        """Returns the collection as a list of dictionaries."""
+        return [resource.to_dict() for resource in self._resource_collection]

mpt_api_client/models/collection.py

@@ -0,0 +1,56 @@
+import math
+from dataclasses import dataclass, field
+from typing import Self
+
+from httpx import Response
+
+
+@dataclass
+class Pagination:
+    """Provides pagination information."""
+
+    limit: int = 0
+    offset: int = 0
+    total: int = 0
+
+    def has_next(self) -> bool:
+        """Returns True if there is a next page."""
+        return self.num_page() + 1 < self.total_pages()
+
+    def num_page(self) -> int:
+        """Returns the current page number starting the first page as 0."""
+        if self.limit == 0:
+            return 0
+        return self.offset // self.limit
+
+    def total_pages(self) -> int:
+        """Returns the total number of pages."""
+        if self.limit == 0:
+            return 0
+        return math.ceil(self.total / self.limit)
+
+    def next_offset(self) -> int:
+        """Returns the next offset as an integer for the next page."""
+        return self.offset + self.limit
+
+
+@dataclass
+class Meta:
+    """Provides meta-information about the pagination, ignored fields and the response."""
+
+    response: Response
+    pagination: Pagination = field(default_factory=Pagination)
+    ignored: list[str] = field(default_factory=list)
+
+    @classmethod
+    def from_response(cls, response: Response) -> Self:
+        """Creates a meta object from response."""
+        meta_data = response.json().get("$meta", {})
+        if not isinstance(meta_data, dict):
+            raise TypeError("Response $meta must be a dict.")
+
+        return cls(
+            ignored=meta_data.get("ignored", []),
+            pagination=Pagination(**meta_data.get("pagination", {})),
+            response=response,
+        )

mpt_api_client/models/meta.py

@@ -0,0 +1,44 @@
+from typing import Any, ClassVar, Self, override
+
+from box import Box
+from httpx import Response
+
+from mpt_api_client.models.meta import Meta
+from mpt_api_client.models.types import BaseResource, ResourceData
+
+
+class Resource(BaseResource):
+    """Provides a resource to interact with api data using fluent interfaces."""
+
+    _data_key: ClassVar[str] = "data"
+    _safe_attributes: ClassVar[list[str]] = ["meta", "_resource_data"]
+
+    def __init__(self, resource_data: ResourceData | None = None, meta: Meta | None = None) -> None:
+        self.meta = meta
+        self._resource_data = Box(resource_data or {}, camel_killer_box=True, default_box=False)
+
+    def __getattr__(self, attribute: str) -> Box | Any:
+        """Returns the resource data."""
+        return self._resource_data.__getattr__(attribute)  # type: ignore[no-untyped-call]
+
+    @override
+    def __setattr__(self, attribute: str, attribute_value: Any) -> None:
+        """Sets the resource data."""
+        if attribute in self._safe_attributes:
+            object.__setattr__(self, attribute, attribute_value)
+            return
+
+        self._resource_data.__setattr__(attribute, attribute_value)  # type: ignore[no-untyped-call]
+
+    @classmethod
+    def from_response(cls, response: Response) -> Self:
+        """Creates a Resource from a response."""
+        response_data = response.json().get(cls._data_key)
+        if not isinstance(response_data, dict):
+            raise TypeError("Response data must be a dict.")
+        meta = Meta.from_response(response)
+        return cls(response_data, meta)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Returns the resource as a dictionary."""
+        return self._resource_data.to_dict()

mpt_api_client/models/resource.py

@@ -0,0 +1,37 @@
+from abc import ABC, abstractmethod
+from typing import Any, Self, TypeVar
+
+from httpx import Response
+
+ResourceData = dict[str, Any]
+ResourceModel = TypeVar("ResourceModel", bound="BaseResource")
+
+
+class BaseResource(ABC):
+    """Provides a base resource to interact with api data using fluent interfaces."""
+
+    @classmethod
+    @abstractmethod
+    def from_response(cls, response: Response) -> Self:
+        """Creates a Resource from a response."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def to_dict(self) -> dict[str, Any]:
+        """Returns the resource as a dictionary."""
+        raise NotImplementedError
+
+
+class BaseCollection(ABC):
+    """Provides a base collection to interact with api collection data using fluent interfaces."""
+
+    @classmethod
+    @abstractmethod
+    def from_response(cls, response: Response) -> Self:
+        """Creates a collection from a response."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def to_list(self) -> list[dict[str, Any]]:
+        """Returns the collection as a list of dictionaries."""
+        raise NotImplementedError

mpt_api_client/models/types.py

@@ -37,7 +37,6 @@ per-file-ignores =
     WPS110
     # Found `noqa` comments overuse
     WPS402
-
   tests/*:
     # Allow magic strings
     WPS432

setup.cfg

@@ -0,0 +1,30 @@
+import pytest
+
+from mpt_api_client.models import Collection, Resource
+
+
+@pytest.fixture
+def meta_data():
+    return {"pagination": {"limit": 10, "offset": 0, "total": 3}, "ignored": ["field1"]}
+
+
+@pytest.fixture
+def response_collection_data():
+    return [
+        {"id": 1, "user": {"name": "Alice", "surname": "Smith"}, "status": "active"},
+        {"id": 2, "user": {"name": "Bob", "surname": "Johnson"}, "status": "inactive"},
+        {"id": 3, "user": {"name": "Charlie", "surname": "Brown"}, "status": "active"},
+    ]
+
+
+TestCollection = Collection[Resource]
+
+
+@pytest.fixture
+def empty_collection() -> TestCollection:
+    return TestCollection()
+
+
+@pytest.fixture
+def collection(response_collection_data):
+    return TestCollection(response_collection_data)

tests/models/collection/conftest.py

@@ -0,0 +1,21 @@
+from httpx import Response
+
+from mpt_api_client.models.collection import Collection
+from mpt_api_client.models.resource import Resource
+
+
+class ChargeResourceMock(Collection[Resource]):
+    _data_key = "charge"
+
+
+def charge(charge_id, amount) -> dict[str, int]:
+    return {"id": charge_id, "amount": amount}
+
+
+def test_custom_data_key():
+    payload = {"charge": [charge(1, 100), charge(2, 101)]}
+    response = Response(200, json=payload)
+
+    resource = ChargeResourceMock.from_response(response)
+
+    assert resource[0].to_dict() == charge(1, 100)