feat: refactoring get_products pagination based on feedback, refactoring get_orders pagination based on feedback, feat: adding pagination for

 get_order_statuses   feat: updating mock backend for get_order_statuses pagination tests: adding test for get order status pagination  tests: adding tests for order_satuses paginatio

Author: theodorehreuter

src/stapi_fastapi/backends/root_backend.py

@@ -6,15 +6,14 @@
 
 from stapi_fastapi.models.order import (
     Order,
-    OrderCollection,
     OrderStatus,
 )
 
 
 class RootBackend[T: OrderStatus](Protocol):  # pragma: nocover
     async def get_orders(
         self, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[OrderCollection, str]]:
+    ) -> ResultE[tuple[list[Order], str]]:
         """
         Return a list of existing orders and pagination token if applicable
         No pagination will return empty string for token

src/stapi_fastapi/routers/root_router.py

@@ -143,51 +143,50 @@ def get_conformance(self, request: Request) -> Conformance:
     def get_products(
         self, request: Request, next: str | None = None, limit: int = 10
     ) -> ProductsCollection:
+        start = 0
+        product_ids = [*self.product_routers.keys()]
+        end = min(start + limit, len(product_ids))
         try:
-            start = 0
-            product_ids = [*self.product_routers.keys()]
             if next:
                 start = product_ids.index(next)
-            if not product_ids and not next:
-                ProductsCollection(
-                    products=[],
-                    links=[
-                        Link(
-                            href=str(request.url_for(f"{self.name}:list-products")),
-                            rel="self",
-                            type=TYPE_JSON,
-                        )
-                    ],
+        except ValueError as e:
+            logging.exception("An error occurred while retrieving orders")
+            raise NotFoundException(detail="Error finding pagination token") from e
+
+        ids = product_ids[start:end]
+        links = [
+            Link(
+                href=str(request.url_for(f"{self.name}:list-products")),
+                rel="self",
+                type=TYPE_JSON,
+            ),
+        ]
+        if end < len(product_ids):
+            links.append(
+                Link(
+                    href=str(
+                        request.url.include_query_params(
+                            next=self.product_routers[product_ids[end]].product.id
+                        ),
+                    ),
+                    rel="next",
+                    type=TYPE_JSON,
                 )
-            end = start + limit
-            ids = product_ids[start:end]
-            products = [
+            )
+        return ProductsCollection(
+            products=[
                 self.product_routers[product_id].get_product(request)
                 for product_id in ids
-            ]
-            links = [
-                Link(
-                    href=str(request.url_for(f"{self.name}:list-products")),
-                    rel="self",
-                    type=TYPE_JSON,
-                ),
-            ]
-            next = ""
-            if end < len(product_ids):
-                next = self.product_routers[product_ids[end]].product.id
-                updated_url = request.url.include_query_params(next=next)
-                links.append(Link(href=str(updated_url), rel="next", type=TYPE_JSON))
-            return ProductsCollection(products=products, links=links)
-        except ValueError as e:
-            logging.exception(f"An error occurred while retrieving orders: {e}")
-            raise NotFoundException(detail="Error finding pagination token")
+            ],
+            links=links,
+        )
 
     async def get_orders(
         self, request: Request, next: str | None = None, limit: int = 10
     ) -> OrderCollection:
         match await self.backend.get_orders(request, next, limit):
-            case Success((collections, token)):
-                for order in collections:
+            case Success((orders, pagination_token)):
+                for order in orders:
                     order.links.append(
                         Link(
                             href=str(
@@ -199,14 +198,27 @@ async def get_orders(
                             type=TYPE_JSON,
                         )
                     )
-                if token:  # pagination link if backend returns token
-                    updated_url = request.url.include_query_params(next=token)
-                    collections.links.append(
-                        Link(href=str(updated_url), rel="next", type=TYPE_JSON)
+                if pagination_token:
+                    return OrderCollection(
+                        features=orders,
+                        links=[
+                            Link(
+                                href=str(
+                                    request.url.include_query_params(
+                                        next=pagination_token
+                                    )
+                                ),
+                                rel="next",
+                                type=TYPE_JSON,
+                            )
+                        ],
                     )
-                return collections
+                return OrderCollection(features=orders)
             case Failure(e):
-                logging.exception(f"An error occurred while retrieving orders: {e}")
+                logger.error(
+                    "An error occurred while retrieving orders: %s",
+                    traceback.format_exception(e),
+                )
                 if isinstance(e, ValueError):
                     raise NotFoundException(detail="Error finding pagination token")
                 else:
@@ -248,31 +260,46 @@ async def get_order_statuses(
         limit: int = 10,
     ) -> OrderStatuses:
         match await self.backend.get_order_statuses(order_id, request, next, limit):
-            case Success(statuses):
-                return OrderStatuses(
-                    statuses=statuses,
-                    links=[
+            case Success((statuses, pagination_token)):
+                links = [
+                    Link(
+                        href=str(
+                            request.url_for(
+                                f"{self.name}:list-order-statuses",
+                                order_id=order_id,
+                            )
+                        ),
+                        rel="self",
+                        type=TYPE_JSON,
+                    )
+                ]
+                if pagination_token:
+                    links.append(
                         Link(
                             href=str(
-                                request.url_for(
-                                    f"{self.name}:list-order-statuses",
-                                    order_id=order_id,
-                                )
+                                request.url.include_query_params(next=pagination_token)
                             ),
-                            rel="self",
+                            rel="next",
                             type=TYPE_JSON,
                         )
-                    ],
+                    )
+                    return OrderStatuses(statuses=statuses, links=links)
+                return OrderStatuses(
+                    statuses=statuses,
+                    links=links,
                 )
             case Failure(e):
                 logger.error(
                     "An error occurred while retrieving order statuses: %s",
                     traceback.format_exception(e),
                 )
-                raise HTTPException(
-                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                    detail="Error finding Order Statuses",
-                )
+                if isinstance(e, KeyError):
+                    raise NotFoundException(detail="Error finding pagination token")
+                else:
+                    raise HTTPException(
+                        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                        detail="Error finding Order Statuses",
+                    )
             case _:
                 raise AssertionError("Expected code to be unreachable")
 

tests/application.py

@@ -18,7 +18,6 @@
 )
 from stapi_fastapi.models.order import (
     Order,
-    OrderCollection,
     OrderParameters,
     OrderPayload,
     OrderStatus,
@@ -45,30 +44,25 @@ def __init__(self, orders: InMemoryOrderDB) -> None:
 
     async def get_orders(
         self, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[OrderCollection, str]]:
+    ) -> ResultE[tuple[list[Order], str]]:
         """
         Return orders from backend.  Handle pagination/limit if applicable
         """
         try:
             start = 0
             if limit > 100:
                 limit = 100
-
             order_ids = [*self._orders_db._orders.keys()]
 
             if next:
                 start = order_ids.index(next)
-            if not order_ids and not next:  # no data in db
-                return Success((OrderCollection(features=[]), ""))
-
-            end = start + limit
+            end = min(start + limit, len(order_ids))
             ids = order_ids[start:end]
-            feats = [self._orders_db._orders[order_id] for order_id in ids]
+            orders = [self._orders_db._orders[order_id] for order_id in ids]
 
-            next = ""
             if end < len(order_ids):
-                next = self._orders_db._orders[order_ids[end]].id
-            return Success((OrderCollection(features=feats), next))
+                return Success((orders, self._orders_db._orders[order_ids[end]].id))
+            return Success((orders, ""))
         except Exception as e:
             return Failure(e)
 
@@ -90,17 +84,12 @@ async def get_order_statuses(
 
             if next:
                 start = int(next)
-            if not statuses and not next:
-                return Success(([], ""))
-
-            end = start + limit
+            end = min(start + limit, len(statuses))
             stati = statuses[start:end]
 
-            next = ""
             if end < len(statuses):
-                next = str(end)
-            return Success((stati, next))
-            # return Success(self._orders_db._statuses[order_id])
+                return Success((stati, str(end)))
+            return Success((stati, ""))
         except Exception as e:
             return Failure(e)
 
@@ -217,3 +206,23 @@ class MyOrderParameters(OrderParameters):
 root_router.add_product(product)
 app: FastAPI = FastAPI()
 app.include_router(root_router, prefix="")
+
+TEST_STATUSES = {
+    "test_order_id": [
+        OrderStatus(
+            timestamp=datetime(2025, 1, 14, 2, 21, 48, 466726, tzinfo=timezone.utc),
+            status_code=OrderStatusCode.received,
+            links=[],
+        ),
+        OrderStatus(
+            timestamp=datetime(2025, 1, 15, 5, 20, 48, 466726, tzinfo=timezone.utc),
+            status_code=OrderStatusCode.accepted,
+            links=[],
+        ),
+        OrderStatus(
+            timestamp=datetime(2025, 1, 16, 10, 15, 32, 466726, tzinfo=timezone.utc),
+            status_code=OrderStatusCode.completed,
+            links=[],
+        ),
+    ]
+}

tests/conftest.py

@@ -21,6 +21,7 @@
 from stapi_fastapi.routers.root_router import RootRouter
 
 from .application import (
+    TEST_STATUSES,
     InMemoryOrderDB,
     MockProductBackend,
     MockRootBackend,
@@ -41,6 +42,13 @@ def order_db() -> InMemoryOrderDB:
     return InMemoryOrderDB()
 
 
+@pytest.fixture
+def order_db_statuses() -> InMemoryOrderDB:
+    order_db = InMemoryOrderDB()
+    order_db._statuses = TEST_STATUSES
+    return order_db
+
+
 @pytest.fixture
 def product_backend(order_db: InMemoryOrderDB) -> MockProductBackend:
     return MockProductBackend(order_db)
@@ -51,6 +59,11 @@ def root_backend(order_db: InMemoryOrderDB) -> MockRootBackend:
     return MockRootBackend(order_db)
 
 
+@pytest.fixture
+def root_backend_preloaded(order_db_statuses: InMemoryOrderDB) -> MockRootBackend:
+    return MockRootBackend(order_db_statuses)
+
+
 @pytest.fixture
 def mock_product_test_spotlight(
     product_backend: MockProductBackend, mock_provider: Provider
@@ -118,6 +131,16 @@ def empty_stapi_client(root_backend, base_url: str) -> Iterator[TestClient]:
         yield client
 
 
+@pytest.fixture
+def statuses_client(root_backend_preloaded, base_url: str) -> Iterator[TestClient]:
+    root_router = RootRouter(root_backend_preloaded)
+    app = FastAPI()
+    app.include_router(root_router, prefix="")
+
+    with TestClient(app, base_url=f"{base_url}") as client:
+        yield client
+
+
 @pytest.fixture(scope="session")
 def url_for(base_url: str) -> Iterator[Callable[[str], str]]:
     def with_trailing_slash(value: str) -> str: