Make sure CursorReference values are not coerced and remove relevant logic

Author: febus982

pyproject.toml

@@ -66,6 +66,7 @@ testpaths = [
 
 [tool.mypy]
 files = "sqlalchemy_bind_manager"
+plugins = "pydantic.mypy"
 
 [tool.ruff]
 select = ["E", "F", "I"]

sqlalchemy_bind_manager/_repository/common.py

@@ -1,6 +1,6 @@
 from typing import Generic, List, TypeVar, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, StrictInt, StrictStr
 from pydantic.generics import GenericModel
 
 MODEL = TypeVar("MODEL")
@@ -23,7 +23,7 @@ class PaginatedResult(GenericModel, Generic[MODEL]):
 
 class CursorReference(BaseModel):
     column: str
-    value: str
+    value: Union[StrictStr, StrictInt]
 
 
 class CursorPageInfo(BaseModel):

sqlalchemy_bind_manager/_repository/result_presenters.py

@@ -1,5 +1,5 @@
 from math import ceil
-from typing import Collection, List, Union
+from typing import List, Union
 
 from sqlalchemy import inspect
 
@@ -56,18 +56,11 @@ def build_result(
             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 not isinstance(last_found_cursor_value, type(cursor_reference.value)):
+                raise TypeError(
+                    "Values from CursorReference and results must be of the same type"
+                )
+            has_next_page = last_found_cursor_value >= cursor_reference.value
             if has_next_page:
                 result_items.pop(index)
             has_previous_page = len(result_items) > items_per_page
@@ -77,20 +70,11 @@ def build_result(
             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 not isinstance(first_found_cursor_value, type(cursor_reference.value)):
+                raise TypeError(
+                    "Values from CursorReference and results must be of the same type"
                 )
+            has_previous_page = first_found_cursor_value <= cursor_reference.value
             if has_previous_page:
                 result_items.pop(index)
             has_next_page = len(result_items) > items_per_page
@@ -117,21 +101,20 @@ def build_result(
 class PaginatedResultPresenter:
     @staticmethod
     def build_result(
-        result_items: Collection[MODEL],
+        result_items: List[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
+        has_next_page = bool(_page and _page < total_pages)
+        has_previous_page = bool(_page and _page > 1)
 
         return PaginatedResult(
             items=result_items,

tests/repository/result_presenters/__init__.py

@@ -0,0 +1,13 @@
+from unittest.mock import Mock, patch
+
+import pytest
+
+from sqlalchemy_bind_manager._repository.result_presenters import _pk_from_result_object
+
+
+def test_exception_raised_if_multiple_primary_keys():
+    with patch(
+        "sqlalchemy_bind_manager._repository.result_presenters.inspect",
+        return_value=Mock(primary_key=["1", "2"]),
+    ), pytest.raises(NotImplementedError):
+        _pk_from_result_object("irrelevant")

tests/repository/result_presenters/test_composite_pk.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+
+import pytest
+
+from sqlalchemy_bind_manager._repository import CursorReference
+from sqlalchemy_bind_manager._repository.result_presenters import (
+    CursorPaginatedResultPresenter,
+)
+
+
+@dataclass
+class MyModel:
+    model_id: int
+    name: str
+
+
+@pytest.mark.parametrize(["is_end_cursor"], [(True,), (False,)])
+def test_fails_if_reference_cursor_wrong_type(is_end_cursor):
+    with pytest.raises(TypeError):
+        CursorPaginatedResultPresenter.build_result(
+            result_items=[MyModel(model_id=1, name="test")],
+            total_items_count=10,
+            items_per_page=1,
+            cursor_reference=CursorReference(column="model_id", value="1"),
+            is_end_cursor=is_end_cursor,
+        )

tests/repository/result_presenters/test_cursor_paginated_result_presenter.py

@@ -0,0 +1,9 @@
+from sqlalchemy_bind_manager._repository import CursorReference
+
+
+def test_cursor_reference_doesnt_coerce_values():
+    r = CursorReference(
+        column="column_name",
+        value=10,
+    )
+    assert isinstance(r.value, int)