♻️ Generic pagination and API model fixtures (ITISFoundation#2682)

Author: pcrespov

packages/service-library/src/servicelib/rest_pagination_utils.py renamed to packages/models-library/src/models_library/rest_pagination.py

@@ -1,5 +1,4 @@
-from math import ceil
-from typing import Any, List, Optional
+from typing import Generic, List, Optional, TypeVar
 
 from pydantic import (
     AnyHttpUrl,
@@ -10,7 +9,7 @@
     PositiveInt,
     validator,
 )
-from yarl import URL
+from pydantic.generics import GenericModel
 
 DEFAULT_NUMBER_OF_ITEMS_PER_PAGE = 20
 
@@ -112,10 +111,17 @@ class Config:
         extra = Extra.forbid
 
 
-class PageResponseLimitOffset(BaseModel):
+ItemT = TypeVar("ItemT")
+
+
+class Page(GenericModel, Generic[ItemT]):
+    """
+    Paginated response model of ItemTs
+    """
+
     meta: PageMetaInfoLimitOffset = Field(alias="_meta")
     links: PageLinks = Field(alias="_links")
-    data: List[Any]
+    data: List[ItemT]
 
     @validator("data", pre=True)
     @classmethod
@@ -136,41 +142,12 @@ def check_data_compatible_with_meta(cls, v, values):
             )
         return v
 
-    @classmethod
-    def paginate_data(
-        cls,
-        data: List[Any],
-        request_url: URL,
-        total: int,
-        limit: int,
-        offset: int,
-    ) -> "PageResponseLimitOffset":
-        last_page = ceil(total / limit) - 1
-
-        return PageResponseLimitOffset(
-            _meta=PageMetaInfoLimitOffset(
-                total=total, count=len(data), limit=limit, offset=offset
-            ),
-            _links=PageLinks(
-                self=f"{request_url.update_query({'offset': offset, 'limit': limit})}",
-                first=f"{request_url.update_query({'offset': 0, 'limit': limit})}",
-                prev=f"{request_url.update_query({'offset': max(offset - limit, 0), 'limit': limit})}"
-                if offset > 0
-                else None,
-                next=f"{request_url.update_query({'offset': min(offset + limit, last_page * limit), 'limit': limit})}"
-                if offset < (last_page * limit)
-                else None,
-                last=f"{request_url.update_query({'offset': last_page * limit, 'limit': limit})}",
-            ),
-            data=data,
-        )
-
     class Config:
         extra = Extra.forbid
 
         schema_extra = {
             "examples": [
-                # first page
+                # first page Page[str]
                 {
                     "_meta": {"total": 7, "count": 4, "limit": 4, "offset": 0},
                     "_links": {

packages/models-library/src/models_library/rest_pagination_utils.py

@@ -0,0 +1,89 @@
+from math import ceil
+from typing import Any, Dict, List, Protocol, TypedDict, Union, runtime_checkable
+
+from .rest_pagination import PageLinks, PageMetaInfoLimitOffset
+
+# NOTE: In this repo we use two type of URL-like data structures:
+#  - from yarl (aiohttp-style) and
+#  - from starlette (fastapi-style)
+#
+# Here define protocol to avoid including starlette  or yarl in this librarie's requirements
+# and a helper function below that can handle both protocols at runtime
+#
+
+
+@runtime_checkable
+class _YarlURL(Protocol):
+    def update_query(self, query) -> "_YarlURL":
+        ...
+
+
+class _StarletteURL(Protocol):
+    # SEE starlette.data_structures.URL
+    #  in https://github.com/encode/starlette/blob/master/starlette/datastructures.py#L130
+
+    def replace_query_params(self, **kwargs: Any) -> "_StarletteURL":
+        ...
+
+
+_URLType = Union[_YarlURL, _StarletteURL]
+
+
+def _replace_query(url: _URLType, query: Dict[str, Any]):
+    """This helper function ensures query replacement works with both"""
+    if isinstance(url, _YarlURL):
+        new_url = url.update_query(query)
+    else:
+        new_url = url.replace_query_params(**query)
+    return f"{new_url}"
+
+
+class PageDict(TypedDict):
+    _meta: Any
+    _links: Any
+    data: List[Any]
+
+
+def paginate_data(
+    chunk: List[Any],
+    *,
+    request_url: _URLType,
+    total: int,
+    limit: int,
+    offset: int,
+) -> PageDict:
+    """Builds page-like objects to feed to Page[ItemT] pydantic model class
+
+    Usage:
+
+        obj: PageDict = paginate_data( ... )
+        model = Page[MyModelItem].parse_obj(obj)
+
+    raises ValidationError
+    """
+    last_page = ceil(total / limit) - 1
+
+    return PageDict(
+        _meta=PageMetaInfoLimitOffset(
+            total=total, count=len(chunk), limit=limit, offset=offset
+        ),
+        _links=PageLinks(
+            self=_replace_query(request_url, {"offset": offset, "limit": limit}),
+            first=_replace_query(request_url, {"offset": 0, "limit": limit}),
+            prev=_replace_query(
+                request_url, {"offset": max(offset - limit, 0), "limit": limit}
+            )
+            if offset > 0
+            else None,
+            next=_replace_query(
+                request_url,
+                {"offset": min(offset + limit, last_page * limit), "limit": limit},
+            )
+            if offset < (last_page * limit)
+            else None,
+            last=_replace_query(
+                request_url, {"offset": last_page * limit, "limit": limit}
+            ),
+        ),
+        data=chunk,
+    )

packages/models-library/tests/test_rest_pagination.py

@@ -0,0 +1,50 @@
+from copy import deepcopy
+
+import pytest
+from models_library.rest_pagination import Page, PageMetaInfoLimitOffset
+from pydantic.main import BaseModel
+
+
+@pytest.mark.parametrize("cls_model", [Page[str], PageMetaInfoLimitOffset])
+def test_page_response_limit_offset_models(cls_model: BaseModel):
+    examples = cls_model.Config.schema_extra["examples"]
+
+    for index, example in enumerate(examples):
+        print(f"{index:-^10}:\n", example)
+
+        model_instance = cls_model(**example)
+        assert model_instance
+
+
+def test_invalid_offset():
+    with pytest.raises(ValueError):
+        PageMetaInfoLimitOffset(limit=6, total=5, offset=5, count=2)
+
+
+@pytest.mark.parametrize(
+    "count, offset",
+    [
+        pytest.param(7, 0, id="count bigger than limit"),
+        pytest.param(6, 0, id="count bigger than total"),
+        pytest.param(5, 1, id="count + offset bigger than total"),
+    ],
+)
+def test_invalid_count(count: int, offset: int):
+    with pytest.raises(ValueError):
+        PageMetaInfoLimitOffset(limit=6, total=5, offset=offset, count=count)
+
+
+def test_data_size_does_not_fit_count():
+    example = deepcopy(Page[str].Config.schema_extra["examples"][0])
+    example["_meta"]["count"] = len(example["data"]) - 1
+    with pytest.raises(ValueError):
+        Page[str](**example)
+
+
+def test_empty_data_is_converted_to_list():
+    example = deepcopy(Page[str].Config.schema_extra["examples"][0])
+    example["data"] = None
+    example["_meta"]["count"] = 0
+    model_instance = Page[str](**example)
+    assert model_instance
+    assert model_instance.data == []

packages/models-library/tests/test_rest_pagination_utils.py

@@ -0,0 +1,170 @@
+import pytest
+from models_library.rest_pagination import Page, PageLinks, PageMetaInfoLimitOffset
+from models_library.rest_pagination_utils import PageDict, paginate_data
+from yarl import URL
+
+
+@pytest.mark.parametrize(
+    "base_url",
+    [
+        "http://site.com",
+        "http://site.com/",
+        "http://some/random/url.com",
+        "http://some/random/url.com/",
+        "http://s.s.s.s.subsite.site.com",
+        "http://s.s.s.s.subsite.site.com/",
+        "http://10.0.0.1.nip.io/",
+        "http://10.0.0.1.nip.io:8091/",
+        "http://10.0.0.1.nip.io",
+        "http://10.0.0.1.nip.io:8091",
+    ],
+)
+def test_paginating_data(base_url):
+    # create random data
+    total_number_of_items = 29
+    limit = 9
+    data_chunk = list(range(limit))
+    request_url = URL(f"{base_url}?some=1&random=4&query=true")
+
+    number_of_chunks = total_number_of_items // limit + 1
+    last_chunk_size = total_number_of_items % limit
+    last_chunk_offset = (number_of_chunks - 1) * len(data_chunk)
+
+    # first "call"
+    offset = 0
+    data_obj: PageDict = paginate_data(
+        data_chunk,
+        total=total_number_of_items,
+        limit=limit,
+        offset=offset,
+        request_url=request_url,
+    )
+    assert data_obj
+
+    model_instance = Page[int].parse_obj(data_obj)
+    assert model_instance
+    assert model_instance.meta == PageMetaInfoLimitOffset(
+        total=total_number_of_items, count=len(data_chunk), limit=limit, offset=offset
+    )
+    assert model_instance.links == PageLinks(
+        self=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={offset}&limit={limit}"
+            )
+        ),
+        first=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset=0&limit={limit}"
+            )
+        ),
+        prev=None,
+        next=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={offset+limit}&limit={limit}"
+            )
+        ),
+        last=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={last_chunk_offset}&limit={limit}"
+            )
+        ),
+    )
+
+    # next "call"s
+    for _ in range(1, number_of_chunks - 1):
+        offset += len(data_chunk)
+        assert model_instance.links.next is not None
+
+        data_obj: PageDict = paginate_data(
+            data_chunk,
+            request_url=URL(model_instance.links.next),
+            total=total_number_of_items,
+            limit=limit,
+            offset=offset,
+        )
+
+        model_instance = Page[int].parse_obj(data_obj)
+        assert model_instance
+        assert model_instance.meta == PageMetaInfoLimitOffset(
+            total=total_number_of_items,
+            count=len(data_chunk),
+            limit=limit,
+            offset=offset,
+        )
+        assert model_instance.links == PageLinks(
+            self=str(
+                URL(base_url).with_query(
+                    f"some=1&random=4&query=true&offset={offset}&limit={limit}"
+                )
+            ),
+            first=str(
+                URL(base_url).with_query(
+                    f"some=1&random=4&query=true&offset=0&limit={limit}"
+                )
+            ),
+            prev=str(
+                URL(base_url).with_query(
+                    f"some=1&random=4&query=true&offset={offset-limit}&limit={limit}"
+                )
+            ),
+            next=str(
+                URL(base_url).with_query(
+                    f"some=1&random=4&query=true&offset={offset+limit}&limit={limit}"
+                )
+            ),
+            last=str(
+                URL(base_url).with_query(
+                    f"some=1&random=4&query=true&offset={last_chunk_offset}&limit={limit}"
+                )
+            ),
+        )
+
+    # last "call"
+    #
+    offset += len(data_chunk)
+    data_chunk = data_chunk[:last_chunk_size]
+
+    assert offset == last_chunk_offset
+
+    assert model_instance.links.next is not None
+    data_obj: PageDict = paginate_data(
+        data_chunk,
+        request_url=URL(model_instance.links.next),
+        total=total_number_of_items,
+        limit=limit,
+        offset=offset,
+    )
+    assert data_obj
+
+    model_instance = Page[int].parse_obj(data_obj)
+    assert model_instance
+
+    assert model_instance.meta == PageMetaInfoLimitOffset(
+        total=total_number_of_items,
+        count=len(data_chunk),
+        limit=limit,
+        offset=offset,
+    )
+    assert model_instance.links == PageLinks(
+        self=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={offset}&limit={limit}"
+            )
+        ),
+        first=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset=0&limit={limit}"
+            )
+        ),
+        prev=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={last_chunk_offset - limit}&limit={limit}"
+            )
+        ),
+        next=None,
+        last=str(
+            URL(base_url).with_query(
+                f"some=1&random=4&query=true&offset={last_chunk_offset}&limit={limit}"
+            )
+        ),
+    )