Move paginated presentation logic in separate classes

Author: febus982

sqlalchemy_bind_manager/_repository/async_.py

@@ -28,6 +28,7 @@
     CursorReference,
     PaginatedResult,
 )
+from .result_presenters import CursorPaginatedResultPresenter, PaginatedResultPresenter
 
 
 class SQLAlchemyAsyncRepository(Generic[MODEL], BaseRepository[MODEL], ABC):
@@ -176,11 +177,11 @@ async def paginated_find(
                 x for x in (await session.execute(paginated_stmt)).scalars()
             ]
 
-            return self._build_page_paginated_result(
+            return PaginatedResultPresenter.build_result(
                 result_items=result_items,
                 total_items_count=total_items_count,
                 page=page,
-                items_per_page=items_per_page,
+                items_per_page=self._sanitised_query_limit(items_per_page),
             )
 
     async def cursor_paginated_find(
@@ -223,10 +224,10 @@ async def cursor_paginated_find(
                 x for x in (await session.execute(paginated_stmt)).scalars()
             ] or []
 
-            return self._build_cursor_paginated_result(
+            return CursorPaginatedResultPresenter.build_result(
                 result_items=result_items,
                 total_items_count=total_items_count,
-                items_per_page=items_per_page,
+                items_per_page=self._sanitised_query_limit(items_per_page),
                 cursor_reference=cursor_reference,
                 is_end_cursor=is_end_cursor,
             )

sqlalchemy_bind_manager/_repository/base_repository.py

@@ -1,13 +1,10 @@
 from abc import ABC
 from enum import Enum
 from functools import partial
-from math import ceil
 from typing import (
     Any,
-    Collection,
     Generic,
     Iterable,
-    List,
     Mapping,
     Tuple,
     Type,
@@ -23,11 +20,7 @@
 
 from .common import (
     MODEL,
-    CursorPageInfo,
-    CursorPaginatedResult,
     CursorReference,
-    PageInfo,
-    PaginatedResult,
 )
 
 
@@ -265,140 +258,6 @@ def _cursor_paginated_query(
     def _sanitised_query_limit(self, limit):
         return max(min(limit, self._max_query_limit), 0)
 
-    def _build_page_paginated_result(
-        self,
-        result_items: Collection[MODEL],
-        total_items_count: int,
-        page: int,
-        items_per_page: int,
-    ) -> PaginatedResult:
-
-        _per_page = self._sanitised_query_limit(items_per_page)
-        total_pages = (
-            0
-            if total_items_count == 0 or total_items_count is None
-            else ceil(total_items_count / _per_page)
-        )
-
-        _page = 0 if len(result_items) == 0 else min(page, total_pages)
-        has_next_page = _page and _page < total_pages
-        has_previous_page = _page and _page > 1
-
-        return PaginatedResult(
-            items=result_items,
-            page_info=PageInfo(
-                page=_page,
-                items_per_page=_per_page,
-                total_items=total_items_count,
-                total_pages=total_pages,
-                has_next_page=has_next_page,
-                has_previous_page=has_previous_page,
-            ),
-        )
-
-    def _build_cursor_paginated_result(
-        self,
-        result_items: List[MODEL],
-        total_items_count: int,
-        items_per_page: int,
-        cursor_reference: Union[CursorReference, None],
-        is_end_cursor: bool,
-    ) -> CursorPaginatedResult:
-        """
-        Produces a structured paginated result identifying previous/next pages
-        and slicing results accordingly.
-
-        :param result_items:
-        :param total_items_count:
-        :param items_per_page:
-        :param cursor_reference:
-        :param is_end_cursor:
-        :return:
-        """
-
-        sanitised_query_limit = self._sanitised_query_limit(items_per_page)
-
-        result_structure: CursorPaginatedResult = CursorPaginatedResult(
-            items=result_items,
-            page_info=CursorPageInfo(
-                items_per_page=sanitised_query_limit,
-                total_items=total_items_count,
-            ),
-        )
-        if not result_items:
-            return result_structure
-
-        if not cursor_reference:
-            has_previous_page = False
-            has_next_page = len(result_items) > sanitised_query_limit
-            if has_next_page:
-                result_items = result_items[0:sanitised_query_limit]
-            # TODO: Infer default cursor format from model
-            result_structure.page_info.has_next_page = has_next_page
-            result_structure.page_info.has_previous_page = has_previous_page
-            reference_column = self._model_pk()
-
-        elif is_end_cursor:
-            index = -1
-            reference_column = cursor_reference.column
-            last_found_cursor_value = getattr(result_items[index], reference_column)
-            """
-            Currently we support only numeric or string model values for cursors,
-            but pydantic models (cursor) coerce always the value as string.
-            This mean if the value is not actually string we need to cast to
-            ensure correct ordering is evaluated.
-            e.g.
-            9 < 10 but '9' > '10' 
-            """
-            if isinstance(last_found_cursor_value, str):
-                has_next_page = last_found_cursor_value >= cursor_reference.value
-            else:
-                has_next_page = last_found_cursor_value >= float(cursor_reference.value)
-            if has_next_page:
-                result_items.pop(index)
-            has_previous_page = len(result_items) > sanitised_query_limit
-            if has_previous_page:
-                result_items = result_items[-sanitised_query_limit:]
-        else:
-            index = 0
-            reference_column = cursor_reference.column
-            first_found_cursor_value = getattr(result_items[index], reference_column)
-            """
-            Currently we support only numeric or string model values for cursors,
-            but pydantic models (cursor) coerce always the value as string.
-            This mean if the value is not actually string we need to cast to
-            ensure correct ordering is evaluated.
-            e.g.
-            9 < 10 but '9' > '10' 
-            """
-            if isinstance(first_found_cursor_value, str):
-                has_previous_page = first_found_cursor_value <= cursor_reference.value
-            else:
-                has_previous_page = first_found_cursor_value <= float(
-                    cursor_reference.value
-                )
-            if has_previous_page:
-                result_items.pop(index)
-            has_next_page = len(result_items) > sanitised_query_limit
-            if has_next_page:
-                result_items = result_items[0:sanitised_query_limit]
-
-        result_structure.items = result_items
-        result_structure.page_info.has_next_page = has_next_page
-        result_structure.page_info.has_previous_page = has_previous_page
-
-        if result_items:
-            result_structure.page_info.start_cursor = CursorReference(
-                column=reference_column,
-                value=getattr(result_items[0], reference_column),
-            )
-            result_structure.page_info.end_cursor = CursorReference(
-                column=reference_column,
-                value=getattr(result_items[-1], reference_column),
-            )
-
-        return result_structure
-
     def _model_pk(self) -> str:
         primary_keys = inspect(self._model).primary_key  # type: ignore
         if len(primary_keys) > 1:

sqlalchemy_bind_manager/_repository/result_presenters.py

@@ -0,0 +1,154 @@
+from math import ceil
+from typing import Collection, List, Union
+
+from sqlalchemy import inspect
+
+from .common import (
+    MODEL,
+    CursorPageInfo,
+    CursorPaginatedResult,
+    CursorReference,
+    PageInfo,
+    PaginatedResult,
+)
+
+
+class CursorPaginatedResultPresenter:
+    @staticmethod
+    def build_result(
+        result_items: List[MODEL],
+        total_items_count: int,
+        items_per_page: int,
+        cursor_reference: Union[CursorReference, None],
+        is_end_cursor: bool,
+    ) -> CursorPaginatedResult:
+        """
+        Produces a structured paginated result identifying previous/next pages
+        and slicing results accordingly.
+
+        :param result_items:
+        :param total_items_count:
+        :param items_per_page:
+        :param cursor_reference:
+        :param is_end_cursor:
+        :return:
+        """
+        result_structure: CursorPaginatedResult = CursorPaginatedResult(
+            items=result_items,
+            page_info=CursorPageInfo(
+                items_per_page=items_per_page,
+                total_items=total_items_count,
+            ),
+        )
+        if not result_items:
+            return result_structure
+
+        if not cursor_reference:
+            has_previous_page = False
+            has_next_page = len(result_items) > items_per_page
+            if has_next_page:
+                result_items = result_items[0:items_per_page]
+            result_structure.page_info.has_next_page = has_next_page
+            result_structure.page_info.has_previous_page = has_previous_page
+            reference_column = _pk_from_result_object(result_items[0])
+
+        elif is_end_cursor:
+            index = -1
+            reference_column = cursor_reference.column
+            last_found_cursor_value = getattr(result_items[index], reference_column)
+            """
+            Currently we support only numeric or string model values for cursors,
+            but pydantic models (cursor) coerce always the value as string.
+            This mean if the value is not actually string we need to cast to
+            ensure correct ordering is evaluated.
+            e.g.
+            9 < 10 but '9' > '10' 
+            """
+            if isinstance(last_found_cursor_value, str):
+                has_next_page = last_found_cursor_value >= cursor_reference.value
+            else:
+                has_next_page = last_found_cursor_value >= float(cursor_reference.value)
+            if has_next_page:
+                result_items.pop(index)
+            has_previous_page = len(result_items) > items_per_page
+            if has_previous_page:
+                result_items = result_items[-items_per_page:]
+        else:
+            index = 0
+            reference_column = cursor_reference.column
+            first_found_cursor_value = getattr(result_items[index], reference_column)
+            """
+            Currently we support only numeric or string model values for cursors,
+            but pydantic models (cursor) coerce always the value as string.
+            This mean if the value is not actually string we need to cast to
+            ensure correct ordering is evaluated.
+            e.g.
+            9 < 10 but '9' > '10' 
+            """
+            if isinstance(first_found_cursor_value, str):
+                has_previous_page = first_found_cursor_value <= cursor_reference.value
+            else:
+                has_previous_page = first_found_cursor_value <= float(
+                    cursor_reference.value
+                )
+            if has_previous_page:
+                result_items.pop(index)
+            has_next_page = len(result_items) > items_per_page
+            if has_next_page:
+                result_items = result_items[0:items_per_page]
+
+        result_structure.items = result_items
+        result_structure.page_info.has_next_page = has_next_page
+        result_structure.page_info.has_previous_page = has_previous_page
+
+        if result_items:
+            result_structure.page_info.start_cursor = CursorReference(
+                column=reference_column,
+                value=getattr(result_items[0], reference_column),
+            )
+            result_structure.page_info.end_cursor = CursorReference(
+                column=reference_column,
+                value=getattr(result_items[-1], reference_column),
+            )
+
+        return result_structure
+
+
+class PaginatedResultPresenter:
+    @staticmethod
+    def build_result(
+        result_items: Collection[MODEL],
+        total_items_count: int,
+        page: int,
+        items_per_page: int,
+    ) -> PaginatedResult:
+
+        total_pages = (
+            0
+            if total_items_count == 0 or total_items_count is None
+            else ceil(total_items_count / items_per_page)
+        )
+
+        _page = 0 if len(result_items) == 0 else min(page, total_pages)
+        has_next_page = _page and _page < total_pages
+        has_previous_page = _page and _page > 1
+
+        return PaginatedResult(
+            items=result_items,
+            page_info=PageInfo(
+                page=_page,
+                items_per_page=items_per_page,
+                total_items=total_items_count,
+                total_pages=total_pages,
+                has_next_page=has_next_page,
+                has_previous_page=has_previous_page,
+            ),
+        )
+
+
+def _pk_from_result_object(model) -> str:
+    primary_keys = inspect(type(model)).primary_key  # type: ignore
+    if len(primary_keys) > 1:
+        raise NotImplementedError("Composite primary keys are not supported.")
+
+    return primary_keys[0].name

sqlalchemy_bind_manager/_repository/sync.py

@@ -28,6 +28,7 @@
     CursorReference,
     PaginatedResult,
 )
+from .result_presenters import CursorPaginatedResultPresenter, PaginatedResultPresenter
 
 
 class SQLAlchemyRepository(Generic[MODEL], BaseRepository[MODEL], ABC):
@@ -164,11 +165,11 @@ def paginated_find(
             )
             result_items = [x for x in session.execute(paginated_stmt).scalars()]
 
-            return self._build_page_paginated_result(
+            return PaginatedResultPresenter.build_result(
                 result_items=result_items,
                 total_items_count=total_items_count,
                 page=page,
-                items_per_page=items_per_page,
+                items_per_page=self._sanitised_query_limit(items_per_page),
             )
 
     def cursor_paginated_find(
@@ -210,10 +211,10 @@ def cursor_paginated_find(
             )
             result_items = [x for x in session.execute(paginated_stmt).scalars()]
 
-            return self._build_cursor_paginated_result(
+            return CursorPaginatedResultPresenter.build_result(
                 result_items=result_items,
                 total_items_count=total_items_count,
-                items_per_page=items_per_page,
+                items_per_page=self._sanitised_query_limit(items_per_page),
                 cursor_reference=cursor_reference,
                 is_end_cursor=is_end_cursor,
             )