feat: creating methods to make Links to improve readability feat: making pagination token return Maybe[str] instead of usng empty strings feat: tweaking logic path in endpoint functions to make a single return point tests: updating tests as necessary to accomodating changes

Author: theodorehreuter

src/stapi_fastapi/backends/product_backend.py

@@ -3,6 +3,7 @@
 from typing import Protocol
 
 from fastapi import Request
+from returns.maybe import Maybe
 from returns.result import ResultE
 
 from stapi_fastapi.models.opportunity import Opportunity, OpportunityRequest
@@ -18,9 +19,9 @@ async def search_opportunities(
         request: Request,
         next: str | None,
         limit: int,
-    ) -> ResultE[tuple[list[Opportunity], str]]:
+    ) -> ResultE[tuple[list[Opportunity], Maybe[str]]]:
         """
-        Search for ordering opportunities for the  given search parameters.
+        Search for ordering opportunities for the  given search parameters and return pagination token if applicable.
 
         Backends must validate search constraints and return
         `stapi_fastapi.exceptions.ConstraintsException` if not valid.

src/stapi_fastapi/backends/root_backend.py

@@ -13,7 +13,7 @@
 class RootBackend[T: OrderStatus](Protocol):  # pragma: nocover
     async def get_orders(
         self, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[list[Order], str]]:
+    ) -> ResultE[tuple[list[Order], Maybe[str]]]:
         """
         Return a list of existing orders and pagination token if applicable
         No pagination will return empty string for token
@@ -35,7 +35,7 @@ async def get_order(self, order_id: str, request: Request) -> ResultE[Maybe[Orde
 
     async def get_order_statuses(
         self, order_id: str, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[list[T], str]]:
+    ) -> ResultE[tuple[list[T], Maybe[str]]]:
         """
         Get statuses for order with `order_id`.
 

src/stapi_fastapi/routers/product_router.py

@@ -6,6 +6,7 @@
 
 from fastapi import APIRouter, HTTPException, Request, Response, status
 from geojson_pydantic.geometries import Geometry
+from returns.maybe import Some
 from returns.result import Failure, Success
 
 from stapi_fastapi.constants import TYPE_JSON
@@ -170,36 +171,25 @@ async def search_opportunities(
         """
         Explore the opportunities available for a particular set of constraints
         """
+        links: list[Link] = []
         match await self.product.backend.search_opportunities(
             self, search, request, next, limit
         ):
-            case Success((features, pagination_token)):
-                links = [
+            case Success((features, Some(pagination_token))):
+                links.append(self.order_link(request, "create-order"))
+                links.append(
                     Link(
                         href=str(
-                            request.url_for(
-                                f"{self.root_router.name}:{self.product.id}:create-order",
-                            ),
+                            request.url.remove_query_params(keys=["next", "limit"])
                         ),
-                        rel="create-order",
+                        rel="next",
                         type=TYPE_JSON,
                         method="POST",
-                    ),
-                ]
-                if pagination_token:
-                    links.append(
-                        Link(
-                            href=str(
-                                request.url.include_query_params(next=pagination_token)
-                            ),
-                            rel="next",
-                            type=TYPE_JSON,
-                            method="POST",
-                            body=search,
-                        )
+                        body={"next": pagination_token, "search": search},
                     )
-                    return OpportunityCollection(features=features, links=links)
-                return OpportunityCollection(features=features, links=links)
+                )
+            case Success((features, Nothing)):  # noqa: F841
+                links.append(self.order_link(request, "create-order"))
             case Failure(e) if isinstance(e, ConstraintsException):
                 raise e
             case Failure(e):
@@ -213,6 +203,7 @@ async def search_opportunities(
                 )
             case x:
                 raise AssertionError(f"Expected code to be unreachable {x}")
+        return OpportunityCollection(features=features, links=links)
 
     def get_product_constraints(self: Self) -> JsonSchemaModel:
         """
@@ -255,3 +246,15 @@ async def create_order(
                 )
             case x:
                 raise AssertionError(f"Expected code to be unreachable {x}")
+
+    def order_link(self, request: Request, suffix: str):
+        return Link(
+            href=str(
+                request.url_for(
+                    f"{self.root_router.name}:{self.product.id}:{suffix}",
+                ),
+            ),
+            rel="create-order",
+            type=TYPE_JSON,
+            method="POST",
+        )

src/stapi_fastapi/routers/root_router.py

@@ -42,6 +42,7 @@ def __init__(
         self.conformances = conformances
         self.openapi_endpoint_name = openapi_endpoint_name
         self.docs_endpoint_name = docs_endpoint_name
+        self.product_ids: list = []
 
         # A dict is used to track the product routers so we can ensure
         # idempotentcy in case a product is added multiple times, and also to
@@ -144,29 +145,28 @@ def get_products(
         self, request: Request, next: str | None = None, limit: int = 10
     ) -> ProductsCollection:
         start = 0
-        product_ids = [*self.product_routers.keys()]
         try:
             if next:
-                start = product_ids.index(next)
+                start = self.product_ids.index(next)
         except ValueError:
-            logging.exception("An error occurred while retrieving orders")
-            raise NotFoundException(detail="Error finding pagination token") from None
-        end = min(start + limit, len(product_ids))
-        ids = product_ids[start:end]
+            logging.exception("An error occurred while retrieving products")
+            raise NotFoundException(
+                detail="Error finding pagination token for products"
+            ) from None
+        end = start + limit
+        ids = self.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) and end != 0:
+        if end > 0 and end < len(self.product_ids):
             links.append(
                 Link(
                     href=str(
-                        request.url.include_query_params(
-                            next=self.product_routers[product_ids[end]].product.id
-                        ),
+                        request.url.include_query_params(next=self.product_ids[end]),
                     ),
                     rel="next",
                     type=TYPE_JSON,
@@ -183,36 +183,24 @@ def get_products(
     async def get_orders(
         self, request: Request, next: str | None = None, limit: int = 10
     ) -> OrderCollection:
+        # links: list[Link] = []
         match await self.backend.get_orders(request, next, limit):
-            case Success((orders, pagination_token)):
+            case Success((orders, Some(pagination_token))):
                 for order in orders:
-                    order.links.append(
-                        Link(
-                            href=str(
-                                request.url_for(
-                                    f"{self.name}:get-order", order_id=order.id
-                                )
-                            ),
-                            rel="self",
-                            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,
-                            )
-                        ],
+                    order.links.append(self.order_link(request, "get-order", order))
+                links = [
+                    Link(
+                        href=str(
+                            request.url.include_query_params(next=pagination_token)
+                        ),
+                        rel="next",
+                        type=TYPE_JSON,
                     )
-                return OrderCollection(features=orders)
+                ]
+            case Success((orders, Nothing)):  # noqa: F841
+                for order in orders:
+                    order.links.append(self.order_link(request, "get-order", order))
+                links = []
             case Failure(e):
                 logger.error(
                     "An error occurred while retrieving orders: %s",
@@ -227,6 +215,7 @@ async def get_orders(
                     )
             case _:
                 raise AssertionError("Expected code to be unreachable")
+        return OrderCollection(features=orders, links=links)
 
     async def get_order(self: Self, order_id: str, request: Request) -> Order:
         """
@@ -258,34 +247,24 @@ async def get_order_statuses(
         next: str | None = None,
         limit: int = 10,
     ) -> OrderStatuses:
+        links: list[Link] = []
         match await self.backend.get_order_statuses(order_id, request, next, limit):
-            case Success((statuses, pagination_token)):
-                links = [
+            case Success((statuses, Some(pagination_token))):
+                links.append(
+                    self.order_statuses_link(request, "list-order-statuses", order_id)
+                )
+                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,
                     )
-                ]
-                if pagination_token:
-                    links.append(
-                        Link(
-                            href=str(
-                                request.url.include_query_params(next=pagination_token)
-                            ),
-                            rel="next",
-                            type=TYPE_JSON,
-                        )
-                    )
-                    return OrderStatuses(statuses=statuses, links=links)
-                return OrderStatuses(
-                    statuses=statuses,
-                    links=links,
+                )
+            case Success((statuses, Nothing)):  # noqa: F841
+                links.append(
+                    self.order_statuses_link(request, "list-order-statuses", order_id)
                 )
             case Failure(e):
                 logger.error(
@@ -301,12 +280,14 @@ async def get_order_statuses(
                     )
             case _:
                 raise AssertionError("Expected code to be unreachable")
+        return OrderStatuses(statuses=statuses, links=links)
 
     def add_product(self: Self, product: Product, *args, **kwargs) -> None:
         # Give the include a prefix from the product router
         product_router = ProductRouter(product, self, *args, **kwargs)
         self.include_router(product_router, prefix=f"/products/{product.id}")
         self.product_routers[product.id] = product_router
+        self.product_ids = [*self.product_routers.keys()]
 
     def generate_order_href(self: Self, request: Request, order_id: str) -> URL:
         return request.url_for(f"{self.name}:get-order", order_id=order_id)
@@ -331,3 +312,22 @@ def add_order_links(self, order: Order, request: Request):
                 type=TYPE_JSON,
             ),
         )
+
+    def order_link(self, request: Request, link_suffix: str, order: Order):
+        return Link(
+            href=str(request.url_for(f"{self.name}:{link_suffix}", order_id=order.id)),
+            rel="self",
+            type=TYPE_JSON,
+        )
+
+    def order_statuses_link(self, request: Request, link_suffix: str, order_id: str):
+        return Link(
+            href=str(
+                request.url_for(
+                    f"{self.name}:{link_suffix}",
+                    order_id=order_id,
+                )
+            ),
+            rel="self",
+            type=TYPE_JSON,
+        )

tests/application.py

@@ -5,7 +5,7 @@
 
 from fastapi import FastAPI, Request
 from pydantic import BaseModel, Field, model_validator
-from returns.maybe import Maybe
+from returns.maybe import Maybe, Nothing, Some
 from returns.result import Failure, ResultE, Success
 
 from stapi_fastapi.backends.product_backend import ProductBackend
@@ -44,7 +44,7 @@ def __init__(self, orders: InMemoryOrderDB) -> None:
 
     async def get_orders(
         self, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[list[Order], str]]:
+    ) -> ResultE[tuple[list[Order], Maybe[str]]]:
         """
         Return orders from backend.  Handle pagination/limit if applicable
         """
@@ -56,13 +56,15 @@ async def get_orders(
 
             if next:
                 start = order_ids.index(next)
-            end = min(start + limit, len(order_ids))
+            end = start + limit
             ids = order_ids[start:end]
             orders = [self._orders_db._orders[order_id] for order_id in ids]
 
-            if end < len(order_ids) and end != 0:
-                return Success((orders, self._orders_db._orders[order_ids[end]].id))
-            return Success((orders, ""))
+            if end > 0 and end < len(order_ids):
+                return Success(
+                    (orders, Some(self._orders_db._orders[order_ids[end]].id))
+                )
+            return Success((orders, Nothing))
         except Exception as e:
             return Failure(e)
 
@@ -75,7 +77,7 @@ async def get_order(self, order_id: str, request: Request) -> ResultE[Maybe[Orde
 
     async def get_order_statuses(
         self, order_id: str, request: Request, next: str | None, limit: int
-    ) -> ResultE[tuple[list[OrderStatus], str]]:
+    ) -> ResultE[tuple[list[OrderStatus], Maybe[str]]]:
         try:
             start = 0
             if limit > 100:
@@ -84,12 +86,12 @@ async def get_order_statuses(
 
             if next:
                 start = int(next)
-            end = min(start + limit, len(statuses))
+            end = start + limit
             stati = statuses[start:end]
 
-            if end < len(statuses) and end != 0:
-                return Success((stati, str(end)))
-            return Success((stati, ""))
+            if end > 0 and end < len(statuses):
+                return Success((stati, Some(str(end))))
+            return Success((stati, Nothing))
         except Exception as e:
             return Failure(e)
 
@@ -107,21 +109,21 @@ async def search_opportunities(
         request: Request,
         next: str | None,
         limit: int,
-    ) -> ResultE[tuple[list[Opportunity], str]]:
+    ) -> ResultE[tuple[list[Opportunity], Maybe[str]]]:
         try:
             start = 0
             if limit > 100:
                 limit = 100
             if next:
                 start = int(next)
-            end = min(start + limit, len(self._opportunities))
+            end = start + limit
             opportunities = [
                 o.model_copy(update=search.model_dump())
                 for o in self._opportunities[start:end]
             ]
-            if end < len(self._opportunities) and end != 0:
-                return Success((opportunities, str(end)))
-            return Success((opportunities, ""))
+            if end > 0 and end < len(self._opportunities):
+                return Success((opportunities, Some(str(end))))
+            return Success((opportunities, Nothing))
         except Exception as e:
             return Failure(e)