feat: tighened up logic in root_router.py to only add pagination link object if we are returned a token. tests: created simple test implementation to abstract out token handling and generation

Author: theodorehreuter

src/stapi_fastapi/routers/root_router.py

@@ -160,9 +160,9 @@ def get_products(self, request: Request) -> ProductsCollection:
         )
 
     async def get_orders(
-        self, request: Request, next_token: str | None = None, limit: int | None = None
+        self, request: Request, next: str | None = None, limit: int | None = None
     ) -> OrderCollection:
-        match await self.backend.get_orders(request, next_token, limit):
+        match await self.backend.get_orders(request, next, limit):
             case Success((collections, token)):
                 for order in collections:
                     order.links.append(
@@ -176,18 +176,14 @@ async def get_orders(
                             type=TYPE_JSON,
                         )
                     )
-                if next_token:
+                if token:  # only return pagination link if backend returns token
                     query = request.url.components.query
-                    if query:  # check url for params
-                        params = {
-                            param.split("=")[0]: param.split("=")[1]
-                            for param in query.split("&")
-                        }
-                        params["next_token"] = token  # replace old token if exists
-                        updated_url = request.url.replace_query_params(**params)
-                    else:  # add if doesn't exist
-                        updated_url = request.url.include_query_params(token=token)
-
+                    params = {
+                        param.split("=")[0]: param.split("=")[1]
+                        for param in query.split("&")
+                    }
+                    params["next"] = token  # replace/add token
+                    updated_url = request.url.replace_query_params(**params)
                     collections.links.append(
                         Link(href=str(updated_url), rel="next", type=TYPE_JSON)
                     )

tests/application.py

@@ -21,7 +21,6 @@
     OrderCollection,
     OrderParameters,
     OrderPayload,
-    Orders,
     OrderStatus,
     OrderStatusCode,
     OrderStatusPayload,
@@ -45,18 +44,51 @@ def __init__(self, orders: InMemoryOrderDB) -> None:
         self._orders_db: InMemoryOrderDB = orders
 
     async def get_orders(
-        self, request: Request, next_token: str, limit: int
-    ) -> ResultE[Orders]:
+        self, request: Request, next: str | None = None, limit: int | None = None
+    ) -> ResultE[tuple[OrderCollection, str]]:
         """
         Show all orders.
         """
-        return Success(
-            Orders(
-                collection=OrderCollection(
-                    features=list(self._orders_db._orders.values())
-                ),
-                token="a",
+        # it limit does NOT reach the last index in the db list THEN we return token
+
+        # if no limit - no token since getting all records - return no token
+        # backend determines if we return a token
+        if next:  # initial implementation - if given a token, assume we continue pagination and return a new token
+            features = list(self._orders_db._orders.values())
+
+            # get records based on limit
+            def token_processor(next: str) -> tuple[str, int]:
+                """process token to return new token and start loc"""
+                return "new_token", 0
+
+            token, start = token_processor(next)
+            if limit:
+                if start + limit > len(features):
+                    features = features[start:]
+                else:
+                    features = features[start:limit]
+                return Success((OrderCollection(features=features), token))
+            else:  # if no limit with token
+                return Success((OrderCollection(features=features[start:]), ""))
+        else:  # no input token means give us everything and return no token
+            return Success(
+                (OrderCollection(features=list(self._orders_db._orders.values())), "")
             )
+        # need to be agnostic to token here.  If we have MORE records we COULD return - i.e. final index found by limit, then we return a token.  If we return the last record THEN return EMPTY token.
+        # token = ''
+        # if not limit:
+        #     limit = 0
+        # # parse token here and do stuff to get starting index
+        # start_index = 0
+        # end_index = start_index + limit
+        # if not limit:
+        #     collection = OrderCollection(features=list(self._orders_db._orders.values()))
+        # else:
+        #     all_orders = list(self._orders_db._orders.values())
+        #     features = [all_orders[i] for i in indicies if 0 <= i < len(lst)]
+        #     collection =
+        return Success(
+            (OrderCollection(features=list(self._orders_db._orders.values())), "")
         )
 
     async def get_order(self, order_id: str, request: Request) -> ResultE[Maybe[Order]]:

tests/test_order.py

@@ -222,13 +222,13 @@ def test_order_pagination(
     )
 
     # call all orders
-    res = stapi_client.get("/orders", params={"next_token": OLD_TOKEN, "limit": 1})
+    res = stapi_client.get("/orders", params={"next": OLD_TOKEN, "limit": 1})
     checker = res.json()
 
     assert res.status_code == status.HTTP_200_OK
 
     # temp check to make sure token link isn't added to inside collection
-    for link in checker["features"][1]["links"]:
+    for link in checker["features"][0]["links"]:
         assert link["rel"] != "next"
     assert checker["links"] != []