stapi-fastapi

FastAPI routes and functions for building a STAPI server.

API

ProductRouter

Bases: APIRouter

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

class ProductRouter(APIRouter):
    # FIXME ruff is complaining that the init is too complex
    def __init__(  # noqa
        self,
        product: Product,
        root_router: RootRouter,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)

        self.product = product
        self.root_router = root_router
        self.conformances = build_conformances(product, root_router)

        self.add_api_route(
            path="",
            endpoint=self.get_product,
            name=f"{self.root_router.name}:{self.product.id}:{GET_PRODUCT}",
            methods=["GET"],
            summary="Retrieve this product",
            tags=["Products"],
        )

        self.add_api_route(
            path="/conformance",
            endpoint=self.get_product_conformance,
            name=f"{self.root_router.name}:{self.product.id}:{CONFORMANCE}",
            methods=["GET"],
            summary="Get conformance urls for the product",
            tags=["Products"],
        )

        self.add_api_route(
            path="/queryables",
            endpoint=self.get_product_queryables,
            name=f"{self.root_router.name}:{self.product.id}:{GET_QUERYABLES}",
            methods=["GET"],
            summary="Get queryables for the product",
            tags=["Products"],
        )

        self.add_api_route(
            path="/order-parameters",
            endpoint=self.get_product_order_parameters,
            name=f"{self.root_router.name}:{self.product.id}:{GET_ORDER_PARAMETERS}",
            methods=["GET"],
            summary="Get order parameters for the product",
            tags=["Products"],
        )

        # 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: OrderPayload,  # type: ignore
            request: Request,
            response: Response,
        ) -> Order[OrderStatus]:
            return await self.create_order(payload, request, response)

        _create_order.__annotations__["payload"] = OrderPayload[
            self.product.order_parameters  # type: ignore
        ]

        self.add_api_route(
            path="/orders",
            endpoint=_create_order,
            name=f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}",
            methods=["POST"],
            response_class=GeoJSONResponse,
            status_code=status.HTTP_201_CREATED,
            summary="Create an order for the product",
            tags=["Products"],
        )

        if product.supports_opportunity_search or (
            self.product.supports_async_opportunity_search and self.root_router.supports_async_opportunity_search
        ):
            self.add_api_route(
                path="/opportunities",
                endpoint=self.search_opportunities,
                name=f"{self.root_router.name}:{self.product.id}:{SEARCH_OPPORTUNITIES}",
                methods=["POST"],
                response_class=GeoJSONResponse,
                # unknown why mypy can't see the queryables property on Product, ignoring
                response_model=OpportunityCollection[
                    Geometry,
                    self.product.opportunity_properties,  # type: ignore
                ],
                responses={
                    201: {
                        "model": OpportunitySearchRecord,
                        "content": {TYPE_JSON: {}},
                    }
                },
                summary="Search Opportunities for the product",
                tags=["Products"],
            )

        if product.supports_async_opportunity_search and root_router.supports_async_opportunity_search:
            self.add_api_route(
                path="/opportunities/{opportunity_collection_id}",
                endpoint=self.get_opportunity_collection,
                name=f"{self.root_router.name}:{self.product.id}:{GET_OPPORTUNITY_COLLECTION}",
                methods=["GET"],
                response_class=GeoJSONResponse,
                summary="Get an Opportunity Collection by ID",
                tags=["Products"],
            )

    def get_product(self, request: Request) -> ProductPydantic:
        links = [
            Link(
                href=str(
                    request.url_for(
                        f"{self.root_router.name}:{self.product.id}:{GET_PRODUCT}",
                    ),
                ),
                rel="self",
                type=TYPE_JSON,
            ),
            Link(
                href=str(
                    request.url_for(
                        f"{self.root_router.name}:{self.product.id}:{CONFORMANCE}",
                    ),
                ),
                rel="conformance",
                type=TYPE_JSON,
            ),
            Link(
                href=str(
                    request.url_for(
                        f"{self.root_router.name}:{self.product.id}:{GET_QUERYABLES}",
                    ),
                ),
                rel="queryables",
                type=TYPE_JSON,
            ),
            Link(
                href=str(
                    request.url_for(
                        f"{self.root_router.name}:{self.product.id}:{GET_ORDER_PARAMETERS}",
                    ),
                ),
                rel="order-parameters",
                type=TYPE_JSON,
            ),
            Link(
                href=str(
                    request.url_for(
                        f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}",
                    ),
                ),
                rel="create-order",
                type=TYPE_JSON,
                method="POST",
            ),
        ]

        if self.product.supports_opportunity_search or (
            self.product.supports_async_opportunity_search and self.root_router.supports_async_opportunity_search
        ):
            links.append(
                Link(
                    href=str(
                        request.url_for(
                            f"{self.root_router.name}:{self.product.id}:{SEARCH_OPPORTUNITIES}",
                        ),
                    ),
                    rel="opportunities",
                    type=TYPE_JSON,
                ),
            )

        return self.product.with_links(links=links)

    async def search_opportunities(
        self,
        search: OpportunityPayload,
        request: Request,
        response: Response,
        prefer: Prefer | None = Depends(get_prefer),
    ) -> OpportunityCollection | Response:  # type: ignore
        """
        Explore the opportunities available for a particular set of queryables
        """
        # sync
        if not (
            self.root_router.supports_async_opportunity_search and self.product.supports_async_opportunity_search
        ) or (prefer is Prefer.wait and self.product.supports_opportunity_search):
            return await self.search_opportunities_sync(
                search,
                request,
                response,
                prefer,
            )

        # async
        if (
            prefer is None
            or prefer is Prefer.respond_async
            or (prefer is Prefer.wait and not self.product.supports_opportunity_search)
        ):
            return await self.search_opportunities_async(search, request, prefer)

        raise AssertionError("Expected code to be unreachable")

    async def search_opportunities_sync(
        self,
        search: OpportunityPayload,
        request: Request,
        response: Response,
        prefer: Prefer | None,
    ) -> OpportunityCollection:  # type: ignore
        links: list[Link] = []
        match await self.product.search_opportunities(
            self,
            search,
            search.next,
            search.limit,
            request,
        ):
            case Success((features, maybe_pagination_token)):
                links.append(self.order_link(request, search))
                match maybe_pagination_token:
                    case Some(x):
                        links.append(self.pagination_link(request, search, x))
                    case Maybe.empty:
                        pass
            case Failure(e) if isinstance(e, QueryablesError):
                raise e
            case Failure(e):
                logger.error(
                    "An error occurred while searching opportunities: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error searching opportunities",
                )
            case x:
                raise AssertionError(f"Expected code to be unreachable {x}")

        if prefer is Prefer.wait and self.root_router.supports_async_opportunity_search:
            response.headers["Preference-Applied"] = "wait"

        return OpportunityCollection(features=features, links=links)

    async def search_opportunities_async(
        self,
        search: OpportunityPayload,
        request: Request,
        prefer: Prefer | None,
    ) -> JSONResponse:
        match await self.product.search_opportunities_async(self, search, request):
            case Success(search_record):
                search_record.links.append(self.root_router.opportunity_search_record_self_link(search_record, request))
                headers = {}
                headers["Location"] = str(
                    self.root_router.generate_opportunity_search_record_href(request, search_record.id)
                )
                if prefer is not None:
                    headers["Preference-Applied"] = "respond-async"
                return JSONResponse(
                    status_code=201,
                    content=search_record.model_dump(mode="json"),
                    headers=headers,
                )
            case Failure(e) if isinstance(e, QueryablesError):
                raise e
            case Failure(e):
                logger.error(
                    "An error occurred while initiating an asynchronous opportunity search: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error initiating an asynchronous opportunity search",
                )
            case x:
                raise AssertionError(f"Expected code to be unreachable: {x}")

    def get_product_conformance(self) -> Conformance:
        """
        Return conformance urls of a specific product
        """
        return Conformance.model_validate({"conforms_to": self.conformances})

    def get_product_queryables(self) -> JsonSchemaModel:
        """
        Return supported queryables of a specific product
        """
        return self.product.queryables

    def get_product_order_parameters(self) -> JsonSchemaModel:
        """
        Return supported order parameters of a specific product
        """
        return self.product.order_parameters

    async def create_order(self, payload: OrderPayload, request: Request, response: Response) -> Order:  # type: ignore
        """
        Create a new order.
        """
        match await self.product.create_order(
            self,
            payload,
            request,
        ):
            case Success(order):
                order.links.extend(self.root_router.order_links(order, request))
                location = str(self.root_router.generate_order_href(request, order.id))
                response.headers["Location"] = location
                return order  # type: ignore
            case Failure(e) if isinstance(e, QueryablesError):
                raise e
            case Failure(e):
                logger.error(
                    "An error occurred while creating order: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error creating order",
                )
            case x:
                raise AssertionError(f"Expected code to be unreachable {x}")

    def order_link(self, request: Request, opp_req: OpportunityPayload) -> Link:
        return Link(
            href=str(
                request.url_for(
                    f"{self.root_router.name}:{self.product.id}:{CREATE_ORDER}",
                ),
            ),
            rel="create-order",
            type=TYPE_JSON,
            method="POST",
            body=opp_req.search_body(),
        )

    def pagination_link(self, request: Request, opp_req: OpportunityPayload, pagination_token: str) -> Link:
        body = opp_req.body()
        body["next"] = pagination_token
        return Link(
            href=str(request.url),
            rel="next",
            type=TYPE_JSON,
            method="POST",
            body=body,
        )

    async def get_opportunity_collection(
        self, opportunity_collection_id: str, request: Request
    ) -> OpportunityCollection:  # type: ignore
        """
        Fetch an opportunity collection generated by an asynchronous opportunity search.
        """
        match await self.product.get_opportunity_collection(
            self,
            opportunity_collection_id,
            request,
        ):
            case Success(Some(opportunity_collection)):
                opportunity_collection.links.append(
                    Link(
                        href=str(
                            request.url_for(
                                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
            case Success(Maybe.empty):
                raise NotFoundError("Opportunity Collection not found")
            case Failure(e):
                logger.error(
                    "An error occurred while fetching opportunity collection: '%s': %s",
                    opportunity_collection_id,
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error fetching Opportunity Collection",
                )
            case x:
                raise AssertionError(f"Expected code to be unreachable {x}")

create_order(payload, request, response) async

Create a new order.

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

async def create_order(self, payload: OrderPayload, request: Request, response: Response) -> Order:  # type: ignore
    """
    Create a new order.
    """
    match await self.product.create_order(
        self,
        payload,
        request,
    ):
        case Success(order):
            order.links.extend(self.root_router.order_links(order, request))
            location = str(self.root_router.generate_order_href(request, order.id))
            response.headers["Location"] = location
            return order  # type: ignore
        case Failure(e) if isinstance(e, QueryablesError):
            raise e
        case Failure(e):
            logger.error(
                "An error occurred while creating order: %s",
                traceback.format_exception(e),
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Error creating order",
            )
        case x:
            raise AssertionError(f"Expected code to be unreachable {x}")

get_opportunity_collection(opportunity_collection_id, request) async

Fetch an opportunity collection generated by an asynchronous opportunity search.

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

async def get_opportunity_collection(
    self, opportunity_collection_id: str, request: Request
) -> OpportunityCollection:  # type: ignore
    """
    Fetch an opportunity collection generated by an asynchronous opportunity search.
    """
    match await self.product.get_opportunity_collection(
        self,
        opportunity_collection_id,
        request,
    ):
        case Success(Some(opportunity_collection)):
            opportunity_collection.links.append(
                Link(
                    href=str(
                        request.url_for(
                            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
        case Success(Maybe.empty):
            raise NotFoundError("Opportunity Collection not found")
        case Failure(e):
            logger.error(
                "An error occurred while fetching opportunity collection: '%s': %s",
                opportunity_collection_id,
                traceback.format_exception(e),
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Error fetching Opportunity Collection",
            )
        case x:
            raise AssertionError(f"Expected code to be unreachable {x}")

get_product_conformance()

Return conformance urls of a specific product

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

def get_product_conformance(self) -> Conformance:
    """
    Return conformance urls of a specific product
    """
    return Conformance.model_validate({"conforms_to": self.conformances})

get_product_order_parameters()

Return supported order parameters of a specific product

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

def get_product_order_parameters(self) -> JsonSchemaModel:
    """
    Return supported order parameters of a specific product
    """
    return self.product.order_parameters

get_product_queryables()

Return supported queryables of a specific product

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

def get_product_queryables(self) -> JsonSchemaModel:
    """
    Return supported queryables of a specific product
    """
    return self.product.queryables

search_opportunities(search, request, response, prefer=Depends(get_prefer)) async

Explore the opportunities available for a particular set of queryables

Source code in stapi-fastapi/src/stapi_fastapi/routers/product_router.py

async def search_opportunities(
    self,
    search: OpportunityPayload,
    request: Request,
    response: Response,
    prefer: Prefer | None = Depends(get_prefer),
) -> OpportunityCollection | Response:  # type: ignore
    """
    Explore the opportunities available for a particular set of queryables
    """
    # sync
    if not (
        self.root_router.supports_async_opportunity_search and self.product.supports_async_opportunity_search
    ) or (prefer is Prefer.wait and self.product.supports_opportunity_search):
        return await self.search_opportunities_sync(
            search,
            request,
            response,
            prefer,
        )

    # async
    if (
        prefer is None
        or prefer is Prefer.respond_async
        or (prefer is Prefer.wait and not self.product.supports_opportunity_search)
    ):
        return await self.search_opportunities_async(search, request, prefer)

    raise AssertionError("Expected code to be unreachable")

RootRouter

Bases: APIRouter

Source code in stapi-fastapi/src/stapi_fastapi/routers/root_router.py

class RootRouter(APIRouter):
    def __init__(
        self,
        get_orders: GetOrders,
        get_order: GetOrder,
        get_order_statuses: GetOrderStatuses | None = None,  # type: ignore
        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] = [API_CONFORMANCE.core],
        name: str = "root",
        openapi_endpoint_name: str = "openapi",
        docs_endpoint_name: str = "swagger_ui_html",
        *args: Any,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)

        _conformances = set(conformances)

        self._get_orders = get_orders
        self._get_order = get_order
        self.__get_order_statuses = get_order_statuses
        self.__get_opportunity_search_records = get_opportunity_search_records
        self.__get_opportunity_search_record = get_opportunity_search_record
        self.__get_opportunity_search_record_statuses = get_opportunity_search_record_statuses
        self.name = name
        self.openapi_endpoint_name = openapi_endpoint_name
        self.docs_endpoint_name = docs_endpoint_name
        self.product_ids: list[str] = []

        # 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
        # manage clobbering if multiple products with the same product_id are
        # added.
        self.product_routers: dict[str, ProductRouter] = {}

        self.add_api_route(
            "/",
            self.get_root,
            methods=["GET"],
            name=f"{self.name}:{ROOT}",
            tags=["Root"],
        )

        self.add_api_route(
            "/conformance",
            self.get_conformance,
            methods=["GET"],
            name=f"{self.name}:{CONFORMANCE}",
            tags=["Conformance"],
        )

        self.add_api_route(
            "/products",
            self.get_products,
            methods=["GET"],
            name=f"{self.name}:{LIST_PRODUCTS}",
            tags=["Products"],
        )

        self.add_api_route(
            "/orders",
            self.get_orders,
            methods=["GET"],
            name=f"{self.name}:{LIST_ORDERS}",
            response_class=GeoJSONResponse,
            tags=["Orders"],
        )

        self.add_api_route(
            "/orders/{order_id}",
            self.get_order,
            methods=["GET"],
            name=f"{self.name}:{GET_ORDER}",
            response_class=GeoJSONResponse,
            tags=["Orders"],
        )

        if self.get_order_statuses is not None:
            _conformances.add(API_CONFORMANCE.order_statuses)
            self.add_api_route(
                "/orders/{order_id}/statuses",
                self.get_order_statuses,
                methods=["GET"],
                name=f"{self.name}:{LIST_ORDER_STATUSES}",
                tags=["Orders"],
            )

        if self.supports_async_opportunity_search:
            _conformances.add(API_CONFORMANCE.searches_opportunity)
            self.add_api_route(
                "/searches/opportunities",
                self.get_opportunity_search_records,
                methods=["GET"],
                name=f"{self.name}:{LIST_OPPORTUNITY_SEARCH_RECORDS}",
                summary="List all Opportunity Search Records",
                tags=["Opportunities"],
            )

            self.add_api_route(
                "/searches/opportunities/{search_record_id}",
                self.get_opportunity_search_record,
                methods=["GET"],
                name=f"{self.name}:{GET_OPPORTUNITY_SEARCH_RECORD}",
                summary="Get an Opportunity Search Record by ID",
                tags=["Opportunities"],
            )

        if self.__get_opportunity_search_record_statuses is not None:
            _conformances.add(API_CONFORMANCE.searches_opportunity_statuses)
            self.add_api_route(
                "/searches/opportunities/{search_record_id}/statuses",
                self.get_opportunity_search_record_statuses,
                methods=["GET"],
                name=f"{self.name}:{GET_OPPORTUNITY_SEARCH_RECORD_STATUSES}",
                summary="Get an Opportunity Search Record statuses by ID",
                tags=["Opportunities"],
            )

        self.conformances = list(_conformances)

    def get_root(self, request: Request) -> RootResponse:
        links = [
            Link(
                href=str(request.url_for(f"{self.name}:{ROOT}")),
                rel="self",
                type=TYPE_JSON,
            ),
            Link(
                href=str(request.url_for(self.openapi_endpoint_name)),
                rel="service-description",
                type=TYPE_JSON,
            ),
            Link(
                href=str(request.url_for(self.docs_endpoint_name)),
                rel="service-docs",
                type="text/html",
            ),
            Link(
                href=str(request.url_for(f"{self.name}:{CONFORMANCE}")),
                rel="conformance",
                type=TYPE_JSON,
            ),
            Link(
                href=str(request.url_for(f"{self.name}:{LIST_PRODUCTS}")),
                rel="products",
                type=TYPE_JSON,
            ),
            Link(
                href=str(request.url_for(f"{self.name}:{LIST_ORDERS}")),
                rel="orders",
                type=TYPE_GEOJSON,
            ),
        ]

        if self.supports_async_opportunity_search:
            links.append(
                Link(
                    href=str(request.url_for(f"{self.name}:{LIST_OPPORTUNITY_SEARCH_RECORDS}")),
                    rel="opportunity-search-records",
                    type=TYPE_JSON,
                ),
            )

        return RootResponse(
            id="STAPI API",
            conformsTo=self.conformances,
            links=links,
        )

    def get_conformance(self) -> Conformance:
        return Conformance(conforms_to=self.conformances)

    def get_products(self, request: Request, next: str | None = None, limit: int = 10) -> ProductsCollection:
        start = 0
        limit = min(limit, 100)
        try:
            if next:
                start = self.product_ids.index(next)
        except ValueError:
            logger.exception("An error occurred while retrieving products")
            raise NotFoundError(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 > 0 and end < len(self.product_ids):
            links.append(self.pagination_link(request, self.product_ids[end], limit))
        return ProductsCollection(
            products=[self.product_routers[product_id].get_product(request) for product_id in ids],
            links=links,
        )

    async def get_orders(
        self, request: Request, next: str | None = None, limit: int = 10
    ) -> OrderCollection[OrderStatus]:
        links: list[Link] = []
        match await self._get_orders(next, limit, request):
            case Success((orders, maybe_pagination_token)):
                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 Maybe.empty:
                        pass
            case Failure(ValueError()):
                raise NotFoundError(detail="Error finding pagination token")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving orders: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding Orders",
                )
            case _:
                raise AssertionError("Expected code to be unreachable")
        return OrderCollection(features=orders, links=links)

    async def get_order(self, order_id: str, request: Request) -> Order[OrderStatus]:
        """
        Get details for order with `order_id`.
        """
        match await self._get_order(order_id, request):
            case Success(Some(order)):
                order.links.extend(self.order_links(order, request))
                return order  # type: ignore
            case Success(Maybe.empty):
                raise NotFoundError("Order not found")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving order '%s': %s",
                    order_id,
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding Order",
                )
            case _:
                raise AssertionError("Expected code to be unreachable")

    async def get_order_statuses(
        self,
        order_id: str,
        request: Request,
        next: str | None = None,
        limit: int = 10,
    ) -> OrderStatuses:  # type: ignore
        links: list[Link] = []
        match await self._get_order_statuses(order_id, next, limit, request):
            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 Maybe.empty:
                        pass
            case Success(Maybe.empty):
                raise NotFoundError("Order not found")
            case Failure(ValueError()):
                raise NotFoundError("Error finding pagination token")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving order statuses: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding Order Statuses",
                )
            case _:
                raise AssertionError("Expected code to be unreachable")
        return OrderStatuses(statuses=statuses, links=links)

    def add_product(self, product: Product, *args: Any, **kwargs: Any) -> 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, request: Request, order_id: str) -> URL:
        return request.url_for(f"{self.name}:{GET_ORDER}", order_id=order_id)

    def generate_order_statuses_href(self, request: Request, order_id: str) -> URL:
        return request.url_for(f"{self.name}:{LIST_ORDER_STATUSES}", order_id=order_id)

    def order_links(self, order: Order[OrderStatus], request: Request) -> list[Link]:
        return [
            Link(
                href=str(self.generate_order_href(request, order.id)),
                rel="self",
                type=TYPE_GEOJSON,
            ),
            Link(
                href=str(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(
            href=str(
                request.url_for(
                    f"{self.name}:{LIST_ORDER_STATUSES}",
                    order_id=order_id,
                )
            ),
            rel="self",
            type=TYPE_JSON,
        )

    def pagination_link(self, request: Request, pagination_token: str, limit: int) -> Link:
        return Link(
            href=str(request.url.include_query_params(next=pagination_token, limit=limit)),
            rel="next",
            type=TYPE_JSON,
        )

    async def get_opportunity_search_records(
        self, request: Request, next: str | None = None, limit: int = 10
    ) -> OpportunitySearchRecords:
        links: list[Link] = []
        match await self._get_opportunity_search_records(next, limit, request):
            case Success((records, maybe_pagination_token)):
                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 Maybe.empty:
                        pass
            case Failure(ValueError()):
                raise NotFoundError(detail="Error finding pagination token")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving opportunity search records: %s",
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding Opportunity Search Records",
                )
            case _:
                raise AssertionError("Expected code to be unreachable")
        return OpportunitySearchRecords(search_records=records, links=links)

    async def get_opportunity_search_record(self, search_record_id: str, request: Request) -> OpportunitySearchRecord:
        """
        Get the Opportunity Search Record with `search_record_id`.
        """
        match await self._get_opportunity_search_record(search_record_id, request):
            case Success(Some(search_record)):
                search_record.links.append(self.opportunity_search_record_self_link(search_record, request))
                return search_record  # type: ignore
            case Success(Maybe.empty):
                raise NotFoundError("Opportunity Search Record not found")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving opportunity search record '%s': %s",
                    search_record_id,
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding Opportunity Search Record",
                )
            case _:
                raise AssertionError("Expected code to be unreachable")

    async def get_opportunity_search_record_statuses(
        self, search_record_id: str, request: Request
    ) -> list[OpportunitySearchStatus]:
        """
        Get the Opportunity Search Record statuses with `search_record_id`.
        """
        match await self._get_opportunity_search_record_statuses(search_record_id, request):
            case Success(Some(search_record_statuses)):
                return search_record_statuses  # type: ignore
            case Success(Maybe.empty):
                raise NotFoundError("Opportunity Search Record not found")
            case Failure(e):
                logger.error(
                    "An error occurred while retrieving opportunity search record statuses '%s': %s",
                    search_record_id,
                    traceback.format_exception(e),
                )
                raise HTTPException(
                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                    detail="Error finding 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) -> URL:
        return request.url_for(
            f"{self.name}:{GET_OPPORTUNITY_SEARCH_RECORD}",
            search_record_id=search_record_id,
        )

    def opportunity_search_record_self_link(
        self, opportunity_search_record: OpportunitySearchRecord, request: Request
    ) -> Link:
        return Link(
            href=str(self.generate_opportunity_search_record_href(request, opportunity_search_record.id)),
            rel="self",
            type=TYPE_JSON,
        )

    @property
    def _get_order_statuses(self) -> GetOrderStatuses:  # type: ignore
        if not self.__get_order_statuses:
            raise AttributeError("Root router does not support order status history")
        return self.__get_order_statuses

    @property
    def _get_opportunity_search_records(self) -> GetOpportunitySearchRecords:
        if not self.__get_opportunity_search_records:
            raise AttributeError("Root router does not support async opportunity search")
        return self.__get_opportunity_search_records

    @property
    def _get_opportunity_search_record(self) -> GetOpportunitySearchRecord:
        if not self.__get_opportunity_search_record:
            raise AttributeError("Root router does not support async opportunity search")
        return self.__get_opportunity_search_record

    @property
    def _get_opportunity_search_record_statuses(self) -> GetOpportunitySearchRecordStatuses:
        if not self.__get_opportunity_search_record_statuses:
            raise AttributeError("Root router does not support async opportunity search status history")
        return self.__get_opportunity_search_record_statuses

    @property
    def supports_async_opportunity_search(self) -> bool:
        return self.__get_opportunity_search_records is not None and self.__get_opportunity_search_record is not None

get_opportunity_search_record(search_record_id, request) async

Get the Opportunity Search Record with search_record_id.

Source code in stapi-fastapi/src/stapi_fastapi/routers/root_router.py

async def get_opportunity_search_record(self, search_record_id: str, request: Request) -> OpportunitySearchRecord:
    """
    Get the Opportunity Search Record with `search_record_id`.
    """
    match await self._get_opportunity_search_record(search_record_id, request):
        case Success(Some(search_record)):
            search_record.links.append(self.opportunity_search_record_self_link(search_record, request))
            return search_record  # type: ignore
        case Success(Maybe.empty):
            raise NotFoundError("Opportunity Search Record not found")
        case Failure(e):
            logger.error(
                "An error occurred while retrieving opportunity search record '%s': %s",
                search_record_id,
                traceback.format_exception(e),
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Error finding Opportunity Search Record",
            )
        case _:
            raise AssertionError("Expected code to be unreachable")

get_opportunity_search_record_statuses(search_record_id, request) async

Get the Opportunity Search Record statuses with search_record_id.

Source code in stapi-fastapi/src/stapi_fastapi/routers/root_router.py

async def get_opportunity_search_record_statuses(
    self, search_record_id: str, request: Request
) -> list[OpportunitySearchStatus]:
    """
    Get the Opportunity Search Record statuses with `search_record_id`.
    """
    match await self._get_opportunity_search_record_statuses(search_record_id, request):
        case Success(Some(search_record_statuses)):
            return search_record_statuses  # type: ignore
        case Success(Maybe.empty):
            raise NotFoundError("Opportunity Search Record not found")
        case Failure(e):
            logger.error(
                "An error occurred while retrieving opportunity search record statuses '%s': %s",
                search_record_id,
                traceback.format_exception(e),
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Error finding Opportunity Search Record statuses",
            )
        case _:
            raise AssertionError("Expected code to be unreachable")

get_order(order_id, request) async

Get details for order with order_id.

Source code in stapi-fastapi/src/stapi_fastapi/routers/root_router.py

async def get_order(self, order_id: str, request: Request) -> Order[OrderStatus]:
    """
    Get details for order with `order_id`.
    """
    match await self._get_order(order_id, request):
        case Success(Some(order)):
            order.links.extend(self.order_links(order, request))
            return order  # type: ignore
        case Success(Maybe.empty):
            raise NotFoundError("Order not found")
        case Failure(e):
            logger.error(
                "An error occurred while retrieving order '%s': %s",
                order_id,
                traceback.format_exception(e),
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Error finding Order",
            )
        case _:
            raise AssertionError("Expected code to be unreachable")