MPT-12840 Update service for agreements asset

Author: albertsola

mpt_api_client/http/base_service.py

@@ -8,7 +8,7 @@
 from mpt_api_client.rql import RQLQuery
 
 
-class ServiceBase[Client, Model: BaseModel]:
+class ServiceBase[Client, Model: BaseModel]:  # noqa: WPS214
     """Service base with agnostic HTTP client."""
 
     _endpoint: str
@@ -22,11 +22,13 @@ def __init__(
         query_rql: RQLQuery | None = None,
         query_order_by: list[str] | None = None,
         query_select: list[str] | None = None,
+        endpoint_params: dict[str, str] | None = None,
     ) -> None:
         self.http_client = http_client
         self.query_rql: RQLQuery | None = query_rql
         self.query_order_by = query_order_by
         self.query_select = query_select
+        self.endpoint_params = endpoint_params or {}
 
     def clone(self) -> Self:
         """Create a copy of collection client for immutable operations.
@@ -39,30 +41,40 @@ def clone(self) -> Self:
             query_rql=self.query_rql,
             query_order_by=copy.copy(self.query_order_by) if self.query_order_by else None,
             query_select=copy.copy(self.query_select) if self.query_select else None,
+            endpoint_params=self.endpoint_params,
         )
 
-    def build_url(self, query_params: dict[str, Any] | None = None) -> str:  # noqa: WPS210
+    @property
+    def endpoint(self) -> str:
+        """Service endpoint URL."""
+        return self._endpoint.format(**self.endpoint_params)
+
+    def build_url(
+        self,
+        query_params: dict[str, Any] | None = None,
+    ) -> str:  # noqa: WPS210
         """Builds the endpoint URL with all the query parameters.
 
         Returns:
             Partial URL with query parameters.
         """
         query_params = query_params or {}
+        if self.query_order_by:
+            query_params.update({"order": ",".join(self.query_order_by)})
+        if self.query_select:
+            query_params.update({"select": ",".join(self.query_select)})
+
         query_parts = [
             f"{param_key}={param_value}" for param_key, param_value in query_params.items()
         ]
-        if self.query_order_by:
-            str_order_by = ",".join(self.query_order_by)
-            query_parts.append(f"order={str_order_by}")
-        if self.query_select:
-            str_query_select = ",".join(self.query_select)
-            query_parts.append(f"select={str_query_select}")
+
         if self.query_rql:
             query_parts.append(str(self.query_rql))
+
         if query_parts:
             query = "&".join(query_parts)
-            return f"{self._endpoint}?{query}"
-        return self._endpoint
+            return f"{self.endpoint}?{query}"
+        return self.endpoint
 
     def order_by(self, *fields: str) -> Self:
         """Returns new collection with ordering setup.
@@ -109,12 +121,13 @@ def select(self, *fields: str) -> Self:
         new_client.query_select = list(fields)
         return new_client
 
-    def _create_collection(self, response: httpx.Response) -> Collection[Model]:
+    @classmethod
+    def _create_collection(cls, response: httpx.Response) -> Collection[Model]:
         meta = Meta.from_response(response)
         return Collection(
             resources=[
-                self._model_class.new(resource, meta)
-                for resource in response.json().get(self._collection_key)
+                cls._model_class.new(resource, meta)
+                for resource in response.json().get(cls._collection_key)
             ],
             meta=meta,
         )

mpt_api_client/http/service.py

@@ -11,7 +11,7 @@
 from mpt_api_client.models.collection import ResourceList
 
 
-class Service[Model: BaseModel](ServiceBase[HTTPClient, Model]):
+class Service[Model: BaseModel](ServiceBase[HTTPClient, Model]):  # noqa: WPS214
     """Immutable service for RESTful resource collections.
 
     Examples:

setup.cfg

@@ -33,7 +33,6 @@ extend-ignore =
 
 per-file-ignores =
   mpt_api_client/rql/query_builder.py: WPS110 WPS115 WPS210 WPS214
-  mpt_api_client/http/service.py: WPS214
   tests/http/test_async_service.py: WPS204 WPS202
   tests/http/test_service.py: WPS204 WPS202
   tests/*:

tests/http/test_base_service.py

@@ -0,0 +1,101 @@
+import pytest
+
+from mpt_api_client import RQLQuery
+from mpt_api_client.http import Service
+from tests.conftest import DummyModel
+
+
+class EndpointDummyService(Service[DummyModel]):
+    _endpoint = "/api/{version}/test"
+    _model_class = DummyModel
+
+
+@pytest.fixture
+def base_dummy_service(http_client):
+    return EndpointDummyService(http_client=http_client, endpoint_params={"version": "v1"})
+
+
+def test_endpoint(http_client):
+    service = EndpointDummyService(http_client=http_client, endpoint_params={"version": "vLatest"})
+
+    assert service.endpoint_params == {"version": "vLatest"}
+    assert service.endpoint == "/api/vLatest/test"
+
+
+def test_filter(base_dummy_service, filter_status_active):
+    new_collection = base_dummy_service.filter(filter_status_active)
+
+    assert base_dummy_service.query_rql is None
+    assert new_collection != base_dummy_service
+    assert new_collection.query_rql == filter_status_active
+
+
+def test_multiple_filters(base_dummy_service) -> None:
+    filter_query = RQLQuery(status="active")
+    filter_query2 = RQLQuery(name="test")
+
+    new_collection = base_dummy_service.filter(filter_query).filter(filter_query2)
+
+    assert base_dummy_service.query_rql is None
+    assert new_collection.query_rql == filter_query & filter_query2
+
+
+def test_select(base_dummy_service) -> None:
+    new_collection = base_dummy_service.select("agreement", "-product")
+
+    assert base_dummy_service.query_select is None
+    assert new_collection != base_dummy_service
+    assert new_collection.query_select == ["agreement", "-product"]
+
+
+def test_select_exception(base_dummy_service) -> None:
+    with pytest.raises(ValueError):
+        base_dummy_service.select("agreement").select("product")
+
+
+def test_order_by(base_dummy_service):
+    new_collection = base_dummy_service.order_by("created", "-name")
+
+    assert base_dummy_service.query_order_by is None
+    assert new_collection != base_dummy_service
+    assert new_collection.query_order_by == ["created", "-name"]
+
+
+def test_order_by_exception(base_dummy_service):
+    with pytest.raises(
+        ValueError, match=r"Ordering is already set. Cannot set ordering multiple times."
+    ):
+        base_dummy_service.order_by("created").order_by("name")
+
+
+def test_url(base_dummy_service, filter_status_active) -> None:
+    custom_collection = (
+        base_dummy_service.filter(filter_status_active)
+        .select("-audit", "product.agreements", "-product.agreements.product")
+        .order_by("-created", "name")
+    )
+
+    url = custom_collection.build_url()
+
+    assert custom_collection != base_dummy_service
+    assert url == (
+        "/api/v1/test?order=-created,name"
+        "&select=-audit,product.agreements,-product.agreements.product"
+        "&eq(status,active)"
+    )
+
+
+def test_clone(base_dummy_service, filter_status_active) -> None:
+    configured = (
+        base_dummy_service.filter(filter_status_active)
+        .order_by("created", "-name")
+        .select("agreement", "-product")
+    )
+
+    cloned = configured.clone()
+
+    assert cloned is not configured
+    assert isinstance(cloned, configured.__class__)
+    assert cloned.http_client is configured.http_client
+    assert str(cloned.query_rql) == str(configured.query_rql)
+    assert cloned.endpoint_params == configured.endpoint_params

tests/http/test_service.py

@@ -4,7 +4,6 @@
 import pytest
 import respx
 
-from mpt_api_client.rql import RQLQuery
 from tests.conftest import DummyModel
 from tests.http.conftest import DummyService
 
@@ -279,81 +278,3 @@ def test_sync_update_resource(dummy_service):
     request = mock_route.calls[0].request
     assert mock_route.call_count == 1
     assert json.loads(request.content.decode()) == resource_data
-
-
-def test_sync_filter(dummy_service, filter_status_active):
-    new_collection = dummy_service.filter(filter_status_active)
-
-    assert dummy_service.query_rql is None
-    assert new_collection != dummy_service
-    assert new_collection.query_rql == filter_status_active
-
-
-def test_sync_multiple_filters(dummy_service) -> None:
-    filter_query = RQLQuery(status="active")
-    filter_query2 = RQLQuery(name="test")
-
-    new_collection = dummy_service.filter(filter_query).filter(filter_query2)
-
-    assert dummy_service.query_rql is None
-    assert new_collection.query_rql == filter_query & filter_query2
-
-
-def test_sync_select(dummy_service) -> None:
-    new_collection = dummy_service.select("agreement", "-product")
-
-    assert dummy_service.query_select is None
-    assert new_collection != dummy_service
-    assert new_collection.query_select == ["agreement", "-product"]
-
-
-def test_sync_select_exception(dummy_service) -> None:
-    with pytest.raises(ValueError):
-        dummy_service.select("agreement").select("product")
-
-
-def test_sync_order_by(dummy_service):
-    new_collection = dummy_service.order_by("created", "-name")
-
-    assert dummy_service.query_order_by is None
-    assert new_collection != dummy_service
-    assert new_collection.query_order_by == ["created", "-name"]
-
-
-def test_sync_order_by_exception(dummy_service):
-    with pytest.raises(
-        ValueError, match=r"Ordering is already set. Cannot set ordering multiple times."
-    ):
-        dummy_service.order_by("created").order_by("name")
-
-
-def test_sync_url(dummy_service, filter_status_active) -> None:
-    custom_collection = (
-        dummy_service.filter(filter_status_active)
-        .select("-audit", "product.agreements", "-product.agreements.product")
-        .order_by("-created", "name")
-    )
-
-    url = custom_collection.build_url()
-
-    assert custom_collection != dummy_service
-    assert url == (
-        "/api/v1/test?order=-created,name"
-        "&select=-audit,product.agreements,-product.agreements.product"
-        "&eq(status,active)"
-    )
-
-
-def test_sync_clone(dummy_service, filter_status_active) -> None:
-    configured = (
-        dummy_service.filter(filter_status_active)
-        .order_by("created", "-name")
-        .select("agreement", "-product")
-    )
-
-    cloned = configured.clone()
-
-    assert cloned is not configured
-    assert isinstance(cloned, configured.__class__)
-    assert cloned.http_client is configured.http_client
-    assert str(cloned.query_rql) == str(configured.query_rql)