refactor url_for

Author: Phil Varner

.pre-commit-config.yaml

@@ -7,23 +7,20 @@ repos:
         language: system
         types: [python]
         pass_filenames: false
-        verbose: true
 
       - id: ruff-format
         name: Format with ruff
         entry: uv run ruff format
         language: system
         types: [python]
         pass_filenames: false
-        verbose: true
 
       - id: mypy
         name: Check typing with mypy
         entry: uv run mypy
         language: system
         types: [python]
         pass_filenames: false
-        verbose: true
 
       - id: pymarkdown
         name: Markdownlint

stapi-fastapi/src/stapi_fastapi/routers/base.py

@@ -0,0 +1,13 @@
+from typing import Any
+
+from fastapi import (
+    APIRouter,
+    Request,
+)
+from starlette.datastructures import URL
+
+
+class StapiFastapiBaseRouter(APIRouter):
+    @staticmethod
+    def url_for(request: Request, name: str, /, **path_params: Any) -> URL:
+        return request.url_for(name, **path_params)

stapi-fastapi/src/stapi_fastapi/routers/product_router.py

@@ -5,7 +5,6 @@
 from typing import TYPE_CHECKING, Any
 
 from fastapi import (
-    APIRouter,
     Depends,
     Header,
     HTTPException,
@@ -38,6 +37,7 @@
 from stapi_fastapi.errors import NotFoundError, QueryablesError
 from stapi_fastapi.models.product import Product
 from stapi_fastapi.responses import GeoJSONResponse
+from stapi_fastapi.routers.base import StapiFastapiBaseRouter
 from stapi_fastapi.routers.route_names import (
     CONFORMANCE,
     CREATE_ORDER,
@@ -84,7 +84,7 @@ def build_conformances(product: Product, root_router: RootRouter) -> list[str]:
     return list(conformances)
 
 
-class ProductRouter(APIRouter):
+class ProductRouter(StapiFastapiBaseRouter):
     # FIXME ruff is complaining that the init is too complex
     def __init__(  # noqa
         self,
@@ -199,10 +199,6 @@ async def _create_order(
                 tags=["Products"],
             )
 
-    @staticmethod
-    def url_for(request: Request, name: str, /, **path_params: Any) -> str:
-        return str(request.url_for(name, **path_params))
-
     def get_product(self, request: Request) -> ProductPydantic:
         links = [
             Link(
@@ -411,7 +407,7 @@ def pagination_link(self, request: Request, opp_req: OpportunityPayload, paginat
         body = opp_req.body()
         body["next"] = pagination_token
         return Link(
-            href=str(request.url),
+            href=request.url,
             rel="next",
             type=TYPE_JSON,
             method="POST",

stapi-fastapi/src/stapi_fastapi/routers/root_router.py

@@ -2,7 +2,7 @@
 import traceback
 from typing import Any
 
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import HTTPException, Request, status
 from returns.maybe import Maybe, Some
 from returns.result import Failure, Success
 from stapi_pydantic import (
@@ -18,6 +18,7 @@
     ProductsCollection,
     RootResponse,
 )
+from starlette.datastructures import URL
 
 from stapi_fastapi.backends.root_backend import (
     GetOpportunitySearchRecord,
@@ -32,6 +33,7 @@
 from stapi_fastapi.errors import NotFoundError
 from stapi_fastapi.models.product import Product
 from stapi_fastapi.responses import GeoJSONResponse
+from stapi_fastapi.routers.base import StapiFastapiBaseRouter
 from stapi_fastapi.routers.product_router import ProductRouter
 from stapi_fastapi.routers.route_names import (
     CONFORMANCE,
@@ -48,7 +50,7 @@
 logger = logging.getLogger(__name__)
 
 
-class RootRouter(APIRouter):
+class RootRouter(StapiFastapiBaseRouter):
     def __init__(
         self,
         get_orders: GetOrders,
@@ -170,10 +172,6 @@ def __init__(
 
         self.conformances = list(_conformances)
 
-    @staticmethod
-    def url_for(request: Request, name: str, /, **path_params: Any) -> str:
-        return str(request.url_for(name, **path_params))
-
     def get_root(self, request: Request) -> RootResponse:
         links = [
             Link(
@@ -245,7 +243,7 @@ def get_products(self, request: Request, next: str | None = None, limit: int = 1
             ),
         ]
         if end > 0 and end < len(self.product_ids):
-            links.append(self.pagination_link(request, self.product_ids[end], limit))
+            links.append(self.pagination_link(request, f"{self.name}:{LIST_PRODUCTS}", self.product_ids[end], limit))
         return ProductsCollection(
             products=[self.product_routers[product_id].get_product(request) for product_id in ids],
             links=links,
@@ -261,8 +259,8 @@ async def get_orders(  # noqa: C901
                 for order in orders:
                     order.links.extend(self.order_links(order, request))
                 match maybe_pagination_token:
-                    case Some(x):
-                        links.append(self.pagination_link(request, x, limit))
+                    case Some(next_):
+                        links.append(self.pagination_link(request, f"{self.name}:{LIST_ORDERS}", next_, limit))
                     case Maybe.empty:
                         pass
                 match maybe_orders_count:
@@ -325,8 +323,12 @@ async def get_order_statuses(
             case Success(Some((statuses, maybe_pagination_token))):
                 links.append(self.order_statuses_link(request, order_id))
                 match maybe_pagination_token:
-                    case Some(x):
-                        links.append(self.pagination_link(request, x, limit))
+                    case Some(next_):
+                        links.append(
+                            self.pagination_link(
+                                request, f"{self.name}:{LIST_ORDER_STATUSES}", next_, limit, order_id=order_id
+                            )
+                        )
                     case Maybe.empty:
                         pass
             case Success(Maybe.empty):
@@ -353,10 +355,10 @@ def add_product(self, product: Product, *args: Any, **kwargs: Any) -> None:
         self.product_routers[product.id] = product_router
         self.product_ids = [*self.product_routers.keys()]
 
-    def generate_order_href(self, request: Request, order_id: str) -> str:
+    def generate_order_href(self, request: Request, order_id: str) -> URL:
         return self.url_for(request, f"{self.name}:{GET_ORDER}", order_id=order_id)
 
-    def generate_order_statuses_href(self, request: Request, order_id: str) -> str:
+    def generate_order_statuses_href(self, request: Request, order_id: str) -> URL:
         return self.url_for(request, f"{self.name}:{LIST_ORDER_STATUSES}", order_id=order_id)
 
     def order_links(self, order: Order[OrderStatus], request: Request) -> list[Link]:
@@ -384,13 +386,9 @@ def order_statuses_link(self, request: Request, order_id: str) -> Link:
             type=TYPE_JSON,
         )
 
-    def pagination_link(self, request: Request, pagination_token: str, limit: int) -> Link:
-        href = str(request.url.include_query_params(next=pagination_token, limit=limit)).replace(
-            str(request.url_for(f"{self.name}:{ROOT}")), self.url_for(request, f"{self.name}:{ROOT}"), 1
-        )
-
+    def pagination_link(self, request: Request, name: str, pagination_token: str, limit: int, **kwargs: Any) -> Link:
         return Link(
-            href=href,
+            href=self.url_for(request, name, **kwargs).include_query_params(next=pagination_token, limit=limit),
             rel="next",
             type=TYPE_JSON,
         )
@@ -404,8 +402,12 @@ async def get_opportunity_search_records(
                 for record in records:
                     record.links.append(self.opportunity_search_record_self_link(record, request))
                 match maybe_pagination_token:
-                    case Some(x):
-                        links.append(self.pagination_link(request, x, limit))
+                    case Some(next_):
+                        links.append(
+                            self.pagination_link(
+                                request, f"{self.name}:{LIST_OPPORTUNITY_SEARCH_RECORDS}", next_, limit
+                            )
+                        )
                     case Maybe.empty:
                         pass
             case Failure(ValueError()):
@@ -470,7 +472,7 @@ async def get_opportunity_search_record_statuses(
             case _:
                 raise AssertionError("Expected code to be unreachable")
 
-    def generate_opportunity_search_record_href(self, request: Request, search_record_id: str) -> str:
+    def generate_opportunity_search_record_href(self, request: Request, search_record_id: str) -> URL:
         return self.url_for(
             request,
             f"{self.name}:{GET_OPPORTUNITY_SEARCH_RECORD}",

stapi-pydantic/src/stapi_pydantic/shared.py

@@ -7,6 +7,7 @@
     SerializerFunctionWrapHandler,
     model_serializer,
 )
+from starlette.datastructures import URL
 
 
 class Link(BaseModel):
@@ -22,8 +23,8 @@ class Link(BaseModel):
 
     # redefining init is a hack to get str type to validate for `href`,
     # as str is ultimately coerced into an AnyUrl automatically anyway
-    def __init__(self, href: AnyUrl | str, **kwargs: Any) -> None:
-        super().__init__(href=href, **kwargs)
+    def __init__(self, href: AnyUrl | URL | str, **kwargs: Any) -> None:
+        super().__init__(href=str(href), **kwargs)
 
     # overriding the default serialization to filter None field values from
     # dumped json