pagination tools

Author: pcrespov

packages/common-library/src/common_library/iter_tools.py

@@ -0,0 +1,72 @@
+from collections.abc import Iterable
+from typing import Annotated
+
+from pydantic import BaseModel, ConfigDict, Field, NonNegativeInt, PositiveInt
+
+
+class PageParams(BaseModel):
+    offset_initial: Annotated[NonNegativeInt, Field(frozen=True)] = 0
+    offset_current: NonNegativeInt = 0  # running offset
+    limit: Annotated[PositiveInt, Field(frozen=True)]
+    total_number_of_items: int | None = None
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    @property
+    def offset(self) -> NonNegativeInt:
+        return self.offset_current
+
+    def has_items_left(self) -> bool:
+        return (
+            self.total_number_of_items is None
+            or self.offset_current < self.total_number_of_items
+        )
+
+    def total_number_of_pages(self) -> NonNegativeInt:
+        assert self.total_number_of_items  # nosec
+        num_items = self.total_number_of_items - self.offset_initial
+        return num_items // self.limit + (1 if num_items % self.limit else 0)
+
+
+def iter_pagination_params(
+    offset: NonNegativeInt = 0,
+    limit: PositiveInt = 100,
+    total_number_of_items: NonNegativeInt | None = None,
+) -> Iterable[PageParams]:
+
+    kwargs = {}
+    if total_number_of_items:
+        kwargs["total_number_of_items"] = total_number_of_items
+
+    page_params = PageParams(
+        offset_initial=offset, offset_current=offset, limit=limit, **kwargs
+    )
+
+    assert page_params.offset_current == page_params.offset_initial  # nosec
+
+    total_count_before = page_params.total_number_of_items
+    page_index = 0
+
+    while page_params.has_items_left():
+
+        yield page_params
+
+        if page_params.total_number_of_items is None:
+            msg = "Must be updated at least before the first iteration, i.e. page_args.total_number_of_items = total_count"
+            raise RuntimeError(msg)
+
+        if (
+            total_count_before
+            and total_count_before != page_params.total_number_of_items
+        ):
+            msg = (
+                f"total_number_of_items cannot change on every iteration: before={total_count_before}, now={page_params.total_number_of_items}."
+                "WARNING: the size of the paginated collection might be changing while it is being iterated?"
+            )
+            raise RuntimeError(msg)
+
+        if page_index == 0:
+            total_count_before = page_params.total_number_of_items
+
+        page_params.offset_current += limit
+        assert page_params.offset == page_params.offset_current  # nosec

packages/common-library/src/common_library/pagination_tools.py

@@ -0,0 +1,86 @@
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+# pylint: disable=too-many-arguments
+
+import asyncio
+from collections.abc import Callable
+
+import pytest
+from common_library.pagination_tools import iter_pagination_params
+from pydantic import ValidationError
+
+
+@pytest.fixture
+def all_items() -> list[int]:
+    return list(range(11))
+
+
+@pytest.fixture
+async def get_page(all_items: list[int]) -> Callable:
+    async def _get_page(offset, limit) -> tuple[list[int], int]:
+        await asyncio.sleep(0)
+        return all_items[offset : offset + limit], len(all_items)
+
+    return _get_page
+
+
+@pytest.mark.parametrize("limit", [2, 3, 5])
+@pytest.mark.parametrize("offset", [0, 1, 5])
+async def test_iter_pages_args(
+    limit: int, offset: int, get_page: Callable, all_items: list[int]
+):
+
+    last_page = [None] * limit
+
+    num_items = len(all_items) - offset
+    expected_num_pages = num_items // limit + (1 if num_items % limit else 0)
+
+    num_pages = 0
+    for page_index, page_args in enumerate(iter_pagination_params(offset, limit)):
+
+        page_items, page_args.total_number_of_items = await get_page(
+            page_args.offset_current, page_args.limit
+        )
+
+        assert set(last_page) != set(page_items)
+        last_page = list(page_items)
+
+        # contains sub-sequence
+        assert str(page_items)[1:-1] in str(all_items)[1:-1]
+
+        num_pages = page_index + 1
+
+    assert last_page[-1] == all_items[-1]
+    assert num_pages == expected_num_pages
+
+    assert not page_args.has_items_left()
+    assert page_args.total_number_of_pages() == num_pages
+
+
+@pytest.mark.parametrize("limit", [-1, 0])
+@pytest.mark.parametrize("offset", [-1])
+def test_iter_pages_args_invalid(limit: int, offset: int):
+
+    with pytest.raises(ValidationError):  # noqa: PT012
+        for _ in iter_pagination_params(offset=offset, limit=limit):
+            pass
+
+
+def test_fails_if_total_number_of_items_not_set():
+    with pytest.raises(  # noqa: PT012
+        RuntimeError,
+        match="page_args.total_number_of_items = total_count",
+    ):
+        for _ in iter_pagination_params(limit=2):
+            pass
+
+
+def test_fails_if_total_number_of_items_changes():
+    with pytest.raises(  # noqa: PT012
+        RuntimeError,
+        match="total_number_of_items cannot change on every iteration",
+    ):
+        for page_params in iter_pagination_params(limit=2, total_number_of_items=4):
+            assert page_params.total_number_of_items == 4
+            page_params.total_number_of_items += 1

packages/common-library/tests/test_iter_tools.py

@@ -4,6 +4,7 @@
 
 import arrow
 from aiohttp import web
+from common_library.pagination_tools import iter_pagination_params
 from models_library.basic_types import IDStr
 from models_library.products import ProductName
 from models_library.projects import ProjectID
@@ -145,34 +146,42 @@ async def list_trashed_projects(
     """
     Lists all projects that were trashed until a specific datetime.
     """
-    projects, _ = await _crud_api_read.list_projects_full_depth(
-        app,
-        user_id=user_id,
-        product_name=product_name,
-        trashed=True,
-        tag_ids_list=[],
-        offset=0,
-        limit=100,  # FIXME: this is only the first 100. redo with yield
-        order_by=OrderBy(field=IDStr("trashed"), direction=OrderDirection.ASC),
-        search_by_multi_columns=None,
-        search_by_project_name=None,
-    )
+    trashed_projects: list[ProjectID] = []
+
+    for page_params in iter_pagination_params(limit=100):
+        (
+            projects,
+            page_params.total_number_of_items,
+        ) = await _crud_api_read.list_projects_full_depth(
+            app,
+            user_id=user_id,
+            product_name=product_name,
+            trashed=True,
+            tag_ids_list=[],
+            offset=page_params.offset,
+            limit=page_params.limit,
+            order_by=OrderBy(field=IDStr("trashed"), direction=OrderDirection.ASC),
+            search_by_multi_columns=None,
+            search_by_project_name=None,
+        )
 
-    # NOTE; this can be done at the database level when projects_repo is refactored
-    # by defining a custom trash_filter that permits some flexibility in the filtering options
-    trashed_projects = []
-    for project in projects:
-        trashed_at, trashed_by, trashed_explicitly = _get_trashed_fields(project)
-
-        if (
-            trashed_at
-            and (until_equal_datetime is None or trashed_at < until_equal_datetime)
-            and trashed_by == user_id
-            and trashed_explicitly
-        ):
-            trashed_projects.append(project)
-
-    return [project["uuid"] for project in trashed_projects]
+        # NOTE: post filtering because for the moment, i do not want ot modify the interface
+        # of _crud_api_read.list_projects_full_depth.
+        # This could not be done at the database level when `projects_repo` is refactored
+        # by defining a custom trash_filter that permits some flexibility in the filtering
+        # options
+        for project in projects:
+            trashed_at, trashed_by, trashed_explicitly = _get_trashed_fields(project)
+
+            if (
+                trashed_at
+                and (until_equal_datetime is None or trashed_at < until_equal_datetime)
+                and trashed_by == user_id
+                and trashed_explicitly
+            ):
+                trashed_projects.append(project["uuid"])
+
+    return trashed_projects
 
 
 async def delete_trashed_project(