add typing to order api route

Author: Phil Varner

bin/server.py

@@ -70,7 +70,7 @@ async def create_order(
         """
         Create a new order.
         """
-        order = Order(
+        order = Order[MyOpportunityProperties, MyOrderParameters](
             id=str(uuid4()),
             geometry=payload.geometry,
             properties={
@@ -91,11 +91,11 @@ async def create_order(
         return order
 
 
-class TestSpotlightProperties(OpportunityProperties):
+class MyOpportunityProperties(OpportunityProperties):
     off_nadir: int
 
 
-class TestOrderParameters(OrderParameters):
+class MyOrderParameters(OrderParameters):
     s3_path: str
 
 
@@ -118,8 +118,8 @@ class TestOrderParameters(OrderParameters):
     keywords=["test", "satellite"],
     providers=[provider],
     links=[],
-    constraints=TestSpotlightProperties,
-    order_parameters=TestOrderParameters,
+    constraints=MyOpportunityProperties,
+    order_parameters=MyOrderParameters,
     backend=product_backend,
 )
 

poetry.lock

@@ -14,6 +14,7 @@ fastapi = "^0.115.0"
 pydantic = "^2.9.2"
 geojson-pydantic = "^1.1.1"
 pygeofilter = "^0.2.4"
+cql2 = "^0.3.1"
 
 [tool.poetry.group.dev]
 optional = true

pyproject.toml

@@ -1,4 +1,4 @@
-from typing import Generic, Literal, TypeVar
+from typing import Any, Generic, Literal, TypeVar
 
 from geojson_pydantic import Feature, FeatureCollection
 from geojson_pydantic.geometries import Geometry
@@ -11,11 +11,11 @@
 
 
 class OrderParameters(BaseModel):
-    model_config = ConfigDict(extra="allow")
+    model_config = ConfigDict(extra="forbid")
 
 
-ORP = TypeVar("ORP", bound=OrderParameters)
 OPP = TypeVar("OPP", bound=OpportunityProperties)
+ORP = TypeVar("ORP", bound=OrderParameters)
 
 
 class OrderRequest(BaseModel, Generic[ORP]):
@@ -24,7 +24,7 @@ class OrderRequest(BaseModel, Generic[ORP]):
     # TODO: validate the CQL2 filter?
     filter: CQL2Filter | None = None
 
-    order_parameters: ORP
+    order_parameters: dict[str, Any]
 
     model_config = ConfigDict(strict=True)
 
@@ -41,14 +41,14 @@ class OrderProperties(BaseModel, Generic[OPP, ORP]):
     model_config = ConfigDict(extra="allow")
 
 
-class Order(Feature[Geometry, OrderProperties]):
+class Order(Feature[Geometry, OrderProperties[OPP, ORP]]):
     # We need to enforce that orders have an id defined, as that is required to
     # retrieve them via the API
     id: StrictInt | StrictStr
     type: Literal["Feature"] = "Feature"
     links: list[Link] = Field(default_factory=list)
 
 
-class OrderCollection(FeatureCollection[Order]):
+class OrderCollection(FeatureCollection[Order[OPP, ORP]]):
     type: Literal["FeatureCollection"] = "FeatureCollection"
     links: list[Link] = Field(default_factory=list)

src/stapi_fastapi/models/order.py

@@ -71,6 +71,13 @@ def __init__(
             summary="Get order parameters for the product",
         )
 
+        # This wraps `self.create_order` to explicitly parameterize `OrderRequest`
+        # for this Product. This must be done programmatically instead of with a type
+        # annotation because it's setting the type dynamically instead of statically, and
+        # pydantic needs this type annotation when doing object conversion. This cannot be done
+        # directly to `self.create_order` because doing it there changes
+        # the annotation on every `ProductRouter` instance's `create_order`, not just
+        # this one's.
         async def _create_order(
             payload: OrderRequest,
             request: Request,
@@ -79,7 +86,10 @@ async def _create_order(
             return await self.create_order(payload, request, response)
 
         _create_order.__annotations__["payload"] = OrderRequest[
-            product.order_parameters  # type: ignore
+            self.product.order_parameters  # type: ignore
+        ]
+        _create_order.__annotations__["return"] = Order[
+            self.product.constraints, self.product.order_parameters  # type: ignore
         ]
 
         self.add_api_route(
@@ -88,6 +98,9 @@ async def _create_order(
             name=f"{self.root_router.name}:{self.product.id}:create-order",
             methods=["POST"],
             response_class=GeoJSONResponse,
+            response_model=Order[
+                self.product.constraints, self.product.order_parameters  # type: ignore
+            ],
             status_code=status.HTTP_201_CREATED,
             summary="Create an order for the product",
             tags=["Products"],

src/stapi_fastapi/routers/product_router.py

@@ -9,6 +9,8 @@
 from stapi_fastapi.models.order import Order, OrderCollection, OrderRequest
 from stapi_fastapi.routers.product_router import ProductRouter
 
+from .shared import SpotlightOpportunityProperties, SpotlightOrderParameters
+
 
 class MockOrderDB(dict[int | str, Order]):
     pass
@@ -58,7 +60,7 @@ async def create_order(
         Create a new order.
         """
         if any(allowed == payload for allowed in self._allowed_payloads):
-            order = Order(
+            order = Order[SpotlightOpportunityProperties, SpotlightOrderParameters](
                 id=str(uuid4()),
                 geometry=payload.geometry,
                 properties={

tests/backends.py

@@ -10,7 +10,7 @@
 from stapi_fastapi.models.order import OrderRequest
 
 from .backends import MockProductBackend
-from .shared import SpotlightOrderParameters, find_link
+from .shared import find_link
 
 NOW = datetime.now(UTC)
 START = NOW
@@ -29,7 +29,7 @@ def create_order_allowed_payloads() -> list[OrderRequest]:
                 datetime.fromisoformat("2024-11-15T18:55:33Z"),
             ),
             filter=None,
-            order_parameters=SpotlightOrderParameters(s3_path="BUCKET"),
+            order_parameters={"s3_path": "BUCKET"},
         ),
     ]