refactor use of starlette.datastructures.URL

Author: Phil Varner

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

@@ -47,6 +47,7 @@
     GET_QUERYABLES,
     SEARCH_OPPORTUNITIES,
 )
+from stapi_fastapi.routers.utils import json_link
 
 if TYPE_CHECKING:
     from stapi_fastapi.routers import RootRouter
@@ -201,28 +202,24 @@ async def _create_order(
 
     def get_product(self, request: Request) -> ProductPydantic:
         links = [
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{GET_PRODUCT}"),
                 rel="self",
-                type=TYPE_JSON,
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{CONFORMANCE}"),
                 rel="conformance",
-                type=TYPE_JSON,
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{GET_QUERYABLES}"),
                 rel="queryables",
-                type=TYPE_JSON,
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{GET_ORDER_PARAMETERS}"),
                 rel="order-parameters",
-                type=TYPE_JSON,
             ),
             Link(
-                href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}"),
+                href=str(self.url_for(request, f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}")),
                 rel="create-order",
                 type=TYPE_JSON,
                 method="POST",
@@ -233,10 +230,9 @@ def get_product(self, request: Request) -> ProductPydantic:
             self.product.supports_async_opportunity_search and self.root_router.supports_async_opportunity_search
         ):
             links.append(
-                Link(
+                json_link(
                     href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{SEARCH_OPPORTUNITIES}"),
                     rel="opportunities",
-                    type=TYPE_JSON,
                 ),
             )
 
@@ -396,7 +392,7 @@ async def create_order(self, payload: OrderPayload, request: Request, response:
 
     def order_link(self, request: Request, opp_req: OpportunityPayload) -> Link:
         return Link(
-            href=self.url_for(request, f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}"),
+            href=str(self.url_for(request, f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}")),
             rel="create-order",
             type=TYPE_JSON,
             method="POST",
@@ -407,7 +403,7 @@ def pagination_link(self, request: Request, opp_req: OpportunityPayload, paginat
         body = opp_req.body()
         body["next"] = pagination_token
         return Link(
-            href=request.url,
+            href=str(request.url),
             rel="next",
             type=TYPE_JSON,
             method="POST",
@@ -427,14 +423,13 @@ async def get_opportunity_collection(
         ):
             case Success(Some(opportunity_collection)):
                 opportunity_collection.links.append(
-                    Link(
+                    json_link(
                         href=self.url_for(
                             request,
                             f"{self.root_router.name}:{self.product.id}:{GET_OPPORTUNITY_COLLECTION}",
                             opportunity_collection_id=opportunity_collection_id,
                         ),
                         rel="self",
-                        type=TYPE_JSON,
                     ),
                 )
                 return opportunity_collection  # type: ignore

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

@@ -29,7 +29,7 @@
     GetOrderStatuses,
 )
 from stapi_fastapi.conformance import API as API_CONFORMANCE
-from stapi_fastapi.constants import TYPE_GEOJSON, TYPE_JSON
+from stapi_fastapi.constants import TYPE_GEOJSON
 from stapi_fastapi.errors import NotFoundError
 from stapi_fastapi.models.product import Product
 from stapi_fastapi.responses import GeoJSONResponse
@@ -46,6 +46,7 @@
     LIST_PRODUCTS,
     ROOT,
 )
+from stapi_fastapi.routers.utils import json_link
 
 logger = logging.getLogger(__name__)
 
@@ -174,44 +175,39 @@ def __init__(
 
     def get_root(self, request: Request) -> RootResponse:
         links = [
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.name}:{ROOT}"),
                 rel="self",
-                type=TYPE_JSON,
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, self.openapi_endpoint_name),
                 rel="service-description",
-                type=TYPE_JSON,
             ),
             Link(
-                href=self.url_for(request, self.docs_endpoint_name),
+                href=str(self.url_for(request, self.docs_endpoint_name)),
                 rel="service-docs",
                 type="text/html",
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.name}:{CONFORMANCE}"),
                 rel="conformance",
-                type=TYPE_JSON,
             ),
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.name}:{LIST_PRODUCTS}"),
                 rel="products",
-                type=TYPE_JSON,
             ),
             Link(
-                href=self.url_for(request, f"{self.name}:{LIST_ORDERS}"),
+                href=str(self.url_for(request, f"{self.name}:{LIST_ORDERS}")),
                 rel="orders",
                 type=TYPE_GEOJSON,
             ),
         ]
 
         if self.supports_async_opportunity_search:
             links.append(
-                Link(
+                json_link(
                     href=self.url_for(request, f"{self.name}:{LIST_OPPORTUNITY_SEARCH_RECORDS}"),
                     rel="opportunity-search-records",
-                    type=TYPE_JSON,
                 ),
             )
 
@@ -236,10 +232,9 @@ def get_products(self, request: Request, next: str | None = None, limit: int = 1
         end = start + limit
         ids = self.product_ids[start:end]
         links = [
-            Link(
+            json_link(
                 href=self.url_for(request, f"{self.name}:{LIST_PRODUCTS}"),
                 rel="self",
-                type=TYPE_JSON,
             ),
         ]
         if end > 0 and end < len(self.product_ids):
@@ -364,33 +359,30 @@ def generate_order_statuses_href(self, request: Request, order_id: str) -> URL:
     def order_links(self, order: Order[OrderStatus], request: Request) -> list[Link]:
         return [
             Link(
-                href=self.generate_order_href(request, order.id),
+                href=str(self.generate_order_href(request, order.id)),
                 rel="self",
                 type=TYPE_GEOJSON,
             ),
-            Link(
+            json_link(
                 href=self.generate_order_statuses_href(request, order.id),
                 rel="monitor",
-                type=TYPE_JSON,
             ),
         ]
 
     def order_statuses_link(self, request: Request, order_id: str) -> Link:
-        return Link(
+        return json_link(
             href=self.url_for(
                 request,
                 f"{self.name}:{LIST_ORDER_STATUSES}",
                 order_id=order_id,
             ),
             rel="self",
-            type=TYPE_JSON,
         )
 
     def pagination_link(self, request: Request, name: str, pagination_token: str, limit: int, **kwargs: Any) -> Link:
-        return Link(
+        return json_link(
             href=self.url_for(request, name, **kwargs).include_query_params(next=pagination_token, limit=limit),
             rel="next",
-            type=TYPE_JSON,
         )
 
     async def get_opportunity_search_records(
@@ -482,10 +474,9 @@ def generate_opportunity_search_record_href(self, request: Request, search_recor
     def opportunity_search_record_self_link(
         self, opportunity_search_record: OpportunitySearchRecord, request: Request
     ) -> Link:
-        return Link(
+        return json_link(
             href=self.generate_opportunity_search_record_href(request, opportunity_search_record.id),
             rel="self",
-            type=TYPE_JSON,
         )
 
     @property

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

@@ -0,0 +1,12 @@
+from stapi_pydantic import Link
+from starlette.datastructures import URL
+
+from stapi_fastapi.constants import TYPE_JSON
+
+
+def json_link(href: URL, rel: str) -> Link:
+    return Link(
+        href=str(href),
+        rel=rel,
+        type=TYPE_JSON,
+    )

stapi-pydantic/src/stapi_pydantic/shared.py

@@ -7,7 +7,6 @@
     SerializerFunctionWrapHandler,
     model_serializer,
 )
-from starlette.datastructures import URL
 
 
 class Link(BaseModel):
@@ -23,7 +22,7 @@ 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 | URL | str, **kwargs: Any) -> None:
+    def __init__(self, href: AnyUrl | str, **kwargs: Any) -> None:
         super().__init__(href=str(href), **kwargs)
 
     # overriding the default serialization to filter None field values from