feat: v0.1 conformances

stapi-fastapi/CHANGELOG.md

@@ -21,6 +21,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) a
 - Renamed all exceptions to errors ([#41](https://github.com/stapi-spec/pystapi/pull/41))
 - stapi-fastapi is now using stapi-pydantic models, deduplicating code
 - Product in stapi-fastapi is now subclass of Product from stapi-pydantic
+- How conformances work ([#90](https://github.com/stapi-spec/pystapi/pull/90))
 
 ## [0.6.0] - 2025-02-11
 

stapi-fastapi/src/stapi_fastapi/conformance.py

@@ -1,5 +1,40 @@
+# This is some slightly strange magic to get "static" structures with
+# attributes, to make it pleasant to use in an editor with autocompletion.
+
+import dataclasses
+from dataclasses import dataclass
+
 from stapi_pydantic.constants import STAPI_VERSION
 
-CORE = f"https://stapi.example.com/v{STAPI_VERSION}/core"
-OPPORTUNITIES = f"https://stapi.example.com/v{STAPI_VERSION}/opportunities"
-ASYNC_OPPORTUNITIES = f"https://stapi.example.com/v{STAPI_VERSION}/async-opportunities"
+
+@dataclass(frozen=True)
+class _All:
+    def all(self) -> list[str]:
+        return [getattr(self, field.name) for field in dataclasses.fields(self)]
+
+
+@dataclass(frozen=True)
+class _Api(_All):
+    core: str = f"https://stapi.example.com/v{STAPI_VERSION}/core"
+    order_statuses: str = f"https://stapi.example.com/v{STAPI_VERSION}/order-statuses"
+    searches_opportunity: str = f"https://stapi.example.com/v{STAPI_VERSION}/searches-opportunity"
+    searches_opportunity_statuses: str = f"https://stapi.example.com/v{STAPI_VERSION}/searches-opportunity-statuses"
+
+
+@dataclass(frozen=True)
+class _Product(_All):
+    opportunities: str = f"https://stapi.example.com/v{STAPI_VERSION}/opportunities"
+    opportunities_async: str = f"https://stapi.example.com/v{STAPI_VERSION}/opportunities-async"
+    geojson_point: str = "https://geojson.org/schema/Point.json"
+    geojson_linestring: str = "https://geojson.org/schema/LineString.json"
+    geojson_polygon: str = "https://geojson.org/schema/Polygon.json"
+    geojson_multi_point: str = "https://geojson.org/schema/MultiPoint.json"
+    geojson_multi_polygon: str = "https://geojson.org/schema/MultiPolygon.json"
+    geojson_multi_linestring: str = "https://geojson.org/schema/MultiLineString.json"
+
+
+API = _Api()
+"""API (top level) conformances"""
+
+PRODUCT = _Product()
+"""Product conformances"""

stapi-fastapi/src/stapi_fastapi/models/product.py

@@ -5,6 +5,8 @@
 from stapi_pydantic import OpportunityProperties, OrderParameters, Queryables
 from stapi_pydantic import Product as BaseProduct
 
+from ..conformance import PRODUCT as PRODUCT_CONFORMANCE
+
 if TYPE_CHECKING:
     from stapi_fastapi.backends.product_backend import (
         CreateOrder,
@@ -77,11 +79,15 @@ def get_opportunity_collection(self) -> GetOpportunityCollection:
 
     @property
     def supports_opportunity_search(self) -> bool:
-        return self._search_opportunities is not None
+        return self._search_opportunities is not None and PRODUCT_CONFORMANCE.opportunities in self.conformsTo
 
     @property
     def supports_async_opportunity_search(self) -> bool:
-        return self._search_opportunities_async is not None and self._get_opportunity_collection is not None
+        return (
+            self._search_opportunities_async is not None
+            and self._get_opportunity_collection is not None
+            and PRODUCT_CONFORMANCE.opportunities_async in self.conformsTo
+        )
 
     @property
     def queryables(self) -> type[Queryables]:

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

@@ -33,6 +33,7 @@
     Product as ProductPydantic,
 )
 
+from stapi_fastapi.conformance import PRODUCT as PRODUCT_CONFORMANCE
 from stapi_fastapi.constants import TYPE_JSON
 from stapi_fastapi.errors import NotFoundError, QueryablesError
 from stapi_fastapi.models.product import Product
@@ -67,7 +68,8 @@ def get_prefer(prefer: str | None = Header(None)) -> str | None:
 
 
 class ProductRouter(APIRouter):
-    def __init__(
+    # FIXME ruff is complaining that the init is too complex
+    def __init__(  # noqa
         self,
         product: Product,
         root_router: RootRouter,
@@ -81,6 +83,16 @@ def __init__(
                 f"Product '{product.id}' must support async opportunity search since the root router does",
             )
 
+        product_conformances = PRODUCT_CONFORMANCE.all()
+        has_geosjon = False
+        for conformance in product.conformsTo:
+            if conformance not in product_conformances:
+                raise ValueError(f"{conformance} is not a valid product conformance")
+            elif conformance.startswith("https://geojson.org/schema/"):  # FIXME total hack
+                has_geosjon = True
+        if not has_geosjon:
+            raise ValueError("product conformance does not contain at least one geojson conformance")
+
         self.product = product
         self.root_router = root_router
 

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

@@ -28,7 +28,7 @@
     GetOrders,
     GetOrderStatuses,
 )
-from stapi_fastapi.conformance import ASYNC_OPPORTUNITIES, CORE
+from stapi_fastapi.conformance import API as API_CONFORMANCE
 from stapi_fastapi.constants import TYPE_GEOJSON, TYPE_JSON
 from stapi_fastapi.errors import NotFoundError
 from stapi_fastapi.models.product import Product
@@ -58,7 +58,7 @@ def __init__(
         get_opportunity_search_records: GetOpportunitySearchRecords | None = None,
         get_opportunity_search_record: GetOpportunitySearchRecord | None = None,
         get_opportunity_search_record_statuses: GetOpportunitySearchRecordStatuses | None = None,
-        conformances: list[str] = [CORE],
+        conformances: list[str] = [API_CONFORMANCE.core],
         name: str = "root",
         openapi_endpoint_name: str = "openapi",
         docs_endpoint_name: str = "swagger_ui_html",
@@ -67,13 +67,10 @@ def __init__(
     ) -> None:
         super().__init__(*args, **kwargs)
 
-        if ASYNC_OPPORTUNITIES in conformances and (
-            not get_opportunity_search_records or not get_opportunity_search_record
-        ):
-            raise ValueError(
-                "`get_opportunity_search_records` and `get_opportunity_search_record` "
-                "are required when advertising async opportunity search conformance"
-            )
+        api_conformances = API_CONFORMANCE.all()
+        for conformance in conformances:
+            if conformance not in api_conformances:
+                raise ValueError(f"{conformance} is not a valid API conformance")
 
         self._get_orders = get_orders
         self._get_order = get_order
@@ -143,7 +140,7 @@ def __init__(
             tags=["Orders"],
         )
 
-        if ASYNC_OPPORTUNITIES in conformances:
+        if API_CONFORMANCE.searches_opportunity in conformances:
             self.add_api_route(
                 "/searches/opportunities",
                 self.get_opportunity_search_records,
@@ -162,6 +159,7 @@ def __init__(
                 tags=["Opportunities"],
             )
 
+        if API_CONFORMANCE.searches_opportunity_statuses in conformances:
             self.add_api_route(
                 "/searches/opportunities/{search_record_id}/statuses",
                 self.get_opportunity_search_record_statuses,
@@ -489,7 +487,7 @@ def _get_opportunity_search_record_statuses(self) -> GetOpportunitySearchRecordS
     @property
     def supports_async_opportunity_search(self) -> bool:
         return (
-            ASYNC_OPPORTUNITIES in self.conformances
+            API_CONFORMANCE.searches_opportunity in self.conformances
             and self._get_opportunity_search_records is not None
             and self._get_opportunity_search_record is not None
         )

stapi-fastapi/tests/application.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 from fastapi import FastAPI
-from stapi_fastapi.conformance import CORE, OPPORTUNITIES
+from stapi_fastapi.conformance import API
 from stapi_fastapi.routers.root_router import RootRouter
 
 from tests.backends import (
@@ -35,7 +35,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
     get_order_statuses=mock_get_order_statuses,
     get_opportunity_search_records=mock_get_opportunity_search_records,
     get_opportunity_search_record=mock_get_opportunity_search_record,
-    conformances=[CORE, OPPORTUNITIES],
+    conformances=[API.core],
 )
 root_router.add_product(product_test_spotlight_sync_opportunity)
 root_router.add_product(product_test_satellite_provider_sync_opportunity)

stapi-fastapi/tests/conftest.py

@@ -7,7 +7,7 @@
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
-from stapi_fastapi.conformance import ASYNC_OPPORTUNITIES, CORE, OPPORTUNITIES
+from stapi_fastapi.conformance import API, PRODUCT
 from stapi_fastapi.models.product import (
     Product,
 )
@@ -75,10 +75,11 @@ async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
         get_orders=mock_get_orders,
         get_order=mock_get_order,
         get_order_statuses=mock_get_order_statuses,
-        conformances=[CORE],
+        conformances=[API.core],
     )
 
     for mock_product in mock_products:
+        mock_product.conformsTo = [PRODUCT.opportunities, PRODUCT.opportunities_async, PRODUCT.geojson_point]
         root_router.add_product(mock_product)
 
     app = FastAPI(lifespan=lifespan)
@@ -112,10 +113,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
         get_opportunity_search_records=mock_get_opportunity_search_records,
         get_opportunity_search_record=mock_get_opportunity_search_record,
         get_opportunity_search_record_statuses=mock_get_opportunity_search_record_statuses,
-        conformances=[CORE, OPPORTUNITIES, ASYNC_OPPORTUNITIES],
+        conformances=[
+            API.core,
+            API.searches_opportunity,
+            API.searches_opportunity_statuses,
+        ],
     )
 
     for mock_product in mock_products:
+        mock_product.conformsTo = [PRODUCT.opportunities, PRODUCT.opportunities_async, PRODUCT.geojson_point]
         root_router.add_product(mock_product)
 
     app = FastAPI(lifespan=lifespan)

stapi-fastapi/tests/shared.py

@@ -12,6 +12,7 @@
 from httpx import Response
 from pydantic import BaseModel, Field, model_validator
 from pytest import fail
+from stapi_fastapi.conformance import PRODUCT
 from stapi_fastapi.models.product import Product
 from stapi_pydantic import (
     Opportunity,
@@ -120,6 +121,7 @@ class MyOrderParameters(OrderParameters):
     description="A provider for Test data",
     roles=[ProviderRole.producer],  # Example role
     url="https://test-provider.example.com",  # Must be a valid URL
+    conformsTo=[PRODUCT.geojson_point],
 )
 
 product_test_spotlight = Product(
@@ -137,6 +139,7 @@ class MyOrderParameters(OrderParameters):
     queryables=MyProductQueryables,
     opportunity_properties=MyOpportunityProperties,
     order_parameters=MyOrderParameters,
+    conformsTo=[PRODUCT.geojson_point],
 )
 
 product_test_spotlight_sync_opportunity = Product(
@@ -154,6 +157,7 @@ class MyOrderParameters(OrderParameters):
     queryables=MyProductQueryables,
     opportunity_properties=MyOpportunityProperties,
     order_parameters=MyOrderParameters,
+    conformsTo=[PRODUCT.geojson_point, PRODUCT.opportunities],
 )
 
 
@@ -172,6 +176,7 @@ class MyOrderParameters(OrderParameters):
     queryables=MyProductQueryables,
     opportunity_properties=MyOpportunityProperties,
     order_parameters=MyOrderParameters,
+    conformsTo=[PRODUCT.geojson_point, PRODUCT.opportunities_async],
 )
 
 product_test_spotlight_sync_async_opportunity = Product(
@@ -189,6 +194,7 @@ class MyOrderParameters(OrderParameters):
     queryables=MyProductQueryables,
     opportunity_properties=MyOpportunityProperties,
     order_parameters=MyOrderParameters,
+    conformsTo=[PRODUCT.geojson_point, PRODUCT.opportunities, PRODUCT.opportunities_async],
 )
 
 product_test_satellite_provider_sync_opportunity = Product(
@@ -206,6 +212,7 @@ class MyOrderParameters(OrderParameters):
     queryables=MyProductQueryables,
     opportunity_properties=MyOpportunityProperties,
     order_parameters=MyOrderParameters,
+    conformsTo=[PRODUCT.geojson_point, PRODUCT.opportunities],
 )
 
 

stapi-fastapi/tests/test_conformance.py

@@ -1,6 +1,6 @@
 from fastapi import status
 from fastapi.testclient import TestClient
-from stapi_fastapi.conformance import CORE
+from stapi_fastapi.conformance import API
 
 
 def test_conformance(stapi_client: TestClient) -> None:
@@ -11,4 +11,8 @@ def test_conformance(stapi_client: TestClient) -> None:
 
     body = res.json()
 
-    assert body["conformsTo"] == [CORE]
+    assert body["conformsTo"] == [API.core]
+
+
+def test_all() -> None:
+    assert API.all() == [API.core, API.order_statuses, API.searches_opportunity, API.searches_opportunity_statuses]

stapi-fastapi/tests/test_root.py

@@ -1,6 +1,6 @@
 from fastapi import status
 from fastapi.testclient import TestClient
-from stapi_fastapi.conformance import CORE
+from stapi_fastapi.conformance import API
 
 
 def test_root(stapi_client: TestClient, assert_link) -> None:
@@ -11,7 +11,7 @@ def test_root(stapi_client: TestClient, assert_link) -> None:
 
     body = res.json()
 
-    assert body["conformsTo"] == [CORE]
+    assert body["conformsTo"] == [API.core]
 
     assert_link("GET /", body, "self", "/")
     assert_link("GET /", body, "service-description", "/openapi.json")