feat: add opportunities methods to STAPI Client

pystapi-client/src/pystapi_client/client.py

@@ -3,14 +3,24 @@
 import urllib.parse
 import warnings
 from collections.abc import Callable, Iterable, Iterator
-from typing import (
-    Any,
-)
+from datetime import datetime
+from typing import Any
 
 from httpx import URL, Request
 from httpx._types import TimeoutTypes
 from pydantic import AnyUrl
-from stapi_pydantic import Link, Order, OrderCollection, Product, ProductsCollection
+from stapi_pydantic import (
+    CQL2Filter,
+    Link,
+    Opportunity,
+    OpportunityCollection,
+    OpportunityPayload,
+    Order,
+    OrderCollection,
+    OrderPayload,
+    Product,
+    ProductsCollection,
+)
 
 from pystapi_client.conformance import ConformanceClasses
 from pystapi_client.exceptions import APIError
@@ -237,11 +247,11 @@ def _supports_opportunities(self) -> bool:
     def _supports_async_opportunities(self) -> bool:
         return self.has_conformance(ConformanceClasses.ASYNC_OPPORTUNITIES)
 
-    def get_products(self, limit: int | None = None) -> Iterator[ProductsCollection]:
+    def get_products(self, limit: int | None = None) -> Iterator[Product]:
         """Get all products from this STAPI API
 
         Returns:
-            ProductsCollection: A collection of STAPI Products
+            Iterator[Product]: An iterator of STAPI Products
         """
         products_endpoint = self._get_products_href()
 
@@ -250,11 +260,11 @@ def get_products(self, limit: int | None = None) -> Iterator[ProductsCollection]
         else:
             parameters = {"limit": limit}
 
-        products_collection_iterator = self.stapi_io.get_pages(
-            products_endpoint, parameters=parameters, lookup_key="products"
-        )
+        products_link = Link(href=products_endpoint, method="GET", body=parameters, rel="")
+
+        products_collection_iterator = self.stapi_io.get_pages(link=products_link, lookup_key="products")
         for products_collection in products_collection_iterator:
-            yield ProductsCollection.model_validate(products_collection)
+            yield from ProductsCollection.model_validate(products_collection).products
 
     def get_product(self, product_id: str) -> Product:
         """Get a single product from this STAPI API
@@ -264,34 +274,102 @@ def get_product(self, product_id: str) -> Product:
 
         Returns:
             Product: A STAPI Product
+        """
+        product_endpoint = self._get_products_href(product_id)
+        product_json = self.stapi_io.read_json(endpoint=product_endpoint)
+        return Product.model_validate(product_json)
 
-        Raises:
-            ValueError if product_id does not exist.
+    def get_product_opportunities(
+        self,
+        product_id: str,
+        date_range: tuple[str, str],
+        geometry: dict[str, Any],
+        cql2_filter: CQL2Filter | None = None,  # type: ignore[type-arg]
+        limit: int = 10,
+    ) -> Iterator[Opportunity]:  # type: ignore[type-arg]
+        # TODO Update return type after the pydantic model generic type is fixed
+        """Get all opportunities for a product from this STAPI API
+        Args:
+            product_id: The Product ID to get opportunities for
+            opportunity_parameters: The parameters for the opportunities
+
+        Returns:
+            Iterator[Opportunity]: An iterator of STAPI Opportunities
         """
+        product_opportunities_endpoint = self._get_products_href(product_id, subpath="opportunities")
+
+        opportunity_parameters = OpportunityPayload.model_validate(
+            {
+                "datetime": (
+                    datetime.fromisoformat(date_range[0]),
+                    datetime.fromisoformat(date_range[1]),
+                ),
+                "geometry": geometry,
+                "filter": cql2_filter,
+                "limit": limit,
+            }
+        )
+        opportunities_first_page_link = Link(
+            href=product_opportunities_endpoint, method="POST", body=opportunity_parameters.model_dump(), rel=""
+        )
+        opportunities_first_page_json, next_link = self.stapi_io._get_next_page(
+            opportunities_first_page_link, "features"
+        )
 
-        product_endpoint = self._get_products_href(product_id)
-        product_json = self.stapi_io.read_json(product_endpoint)
+        if opportunities_first_page_json:
+            opportunities_first_page = OpportunityCollection.model_validate(opportunities_first_page_json)  # type:ignore[var-annotated]
+            yield from opportunities_first_page.features
+        else:
+            return
 
-        if product_json is None:
-            raise ValueError(f"Product {product_id} not found")
+        if next_link is None:
+            return
 
-        return Product.model_validate(product_json)
+        product_opportunities_json = self.stapi_io.get_pages(link=next_link, lookup_key="features")
+
+        for opportunity_collection in product_opportunities_json:
+            yield from OpportunityCollection.model_validate(opportunity_collection).features
+
+    def create_product_order(self, product_id: str, order_parameters: OrderPayload) -> Order:  # type: ignore[type-arg]
+        # TODO Update return type after the pydantic model generic type is fixed
+        """Create an order for a product
+
+        Args:
+            product_id: The Product ID to place an order for
+            order_parameters: The parameters for the order
+        """
+        product_order_endpoint = self._get_products_href(product_id, subpath="orders")
+        product_order_json = self.stapi_io.read_json(
+            endpoint=product_order_endpoint, method="POST", parameters=order_parameters.model_dump()
+        )
 
-    def _get_products_href(self, product_id: str | None = None) -> str:
+        return Order.model_validate(product_order_json)
+
+    def _get_products_href(self, product_id: str | None = None, subpath: str | None = None) -> str:
         product_link = self.get_single_link("products")
         if product_link is None:
             raise ValueError("No products link found")
         product_url = URL(str(product_link.href))
+
+        path = None
+
         if product_id is not None:
-            product_url = product_url.copy_with(path=f"{product_url.path}/{product_id}")
+            path = f"{product_url.path}/{product_id}"
+
+        if subpath is not None:
+            path = f"{path}/{subpath}"
+
+        if path is not None:
+            product_url = product_url.copy_with(path=path)
+
         return str(product_url)
 
-    def get_orders(self, limit: int | None = None) -> Iterator[OrderCollection]:  # type: ignore[type-arg]
+    def get_orders(self, limit: int | None = None) -> Iterator[Order]:  # type: ignore[type-arg]
         # TODO Update return type after the pydantic model generic type is fixed
         """Get orders from this STAPI API
 
         Returns:
-            OrderCollection: A collection of STAPI Orders
+            Iterator[Order]: An iterator of STAPI Orders
         """
         orders_endpoint = self._get_orders_href()
 
@@ -300,11 +378,11 @@ def get_orders(self, limit: int | None = None) -> Iterator[OrderCollection]:  #
         else:
             parameters = {"limit": limit}
 
-        orders_collection_iterator = self.stapi_io.get_pages(
-            orders_endpoint, parameters=parameters, lookup_key="features"
-        )
+        orders_link = Link(href=orders_endpoint, method="GET", body=parameters, rel="")
+
+        orders_collection_iterator = self.stapi_io.get_pages(link=orders_link, lookup_key="features")
         for orders_collection in orders_collection_iterator:
-            yield OrderCollection.model_validate(orders_collection)
+            yield from OrderCollection.model_validate(orders_collection).features
 
     def get_order(self, order_id: str) -> Order:  # type: ignore[type-arg]
         # TODO Update return type after the pydantic model generic type is fixed

pystapi-client/src/pystapi_client/stapi_api_io.py

@@ -163,14 +163,28 @@ def read_json(self, endpoint: str, method: str = "GET", parameters: dict[str, An
             The parsed JSON response
         """
         href = urllib.parse.urljoin(str(self.root_url), endpoint)
+
+        if method == "POST" and parameters is None:
+            parameters = {}
+
         text = self._read_text(href, method=method, parameters=parameters)
         return json.loads(text)  # type: ignore[no-any-return]
 
+    def _get_next_page(self, link: Link, lookup_key: str) -> tuple[dict[str, Any] | None, Link | None]:
+        page = self.read_json(str(link.href), method=link.method or "GET", parameters=link.body)
+        next_link = next((link for link in page.get("links", []) if link["rel"] == "next"), None)
+
+        if next_link is not None:
+            next_link = Link.model_validate(next_link)
+
+        if page.get(lookup_key):
+            return page, next_link
+
+        return None, None
+
     def get_pages(
         self,
-        url: str,
-        method: str = "GET",
-        parameters: dict[str, Any] | None = None,
+        link: Link,
         lookup_key: str | None = None,
     ) -> Iterator[dict[str, Any]]:
         """Iterator that yields dictionaries for each page at a STAPI paging
@@ -181,23 +195,17 @@ def get_pages(
         Return:
             dict[str, Any] : JSON content from a single page
         """
-        # TODO update this
-
         if not lookup_key:
             lookup_key = "features"
 
-        page = self.read_json(url, method=method, parameters=parameters)
-        if not (page.get(lookup_key)):
+        first_page, next_link = self._get_next_page(link, lookup_key)
+
+        if first_page is None:
             return None
-        yield page
+        yield first_page
 
-        next_link = next((link for link in page.get("links", []) if link["rel"] == "next"), None)
         while next_link:
-            link = Link.model_validate(next_link)
-            page = self.read_json(str(link.href), method=link.method or "GET")
-            if not (page.get(lookup_key)):
+            next_page, next_link = self._get_next_page(next_link, lookup_key)
+            if next_page is None:
                 return None
-            yield page
-
-            # get the next link and make the next request
-            next_link = next((link for link in page.get("links", []) if link["rel"] == "next"), None)
+            yield next_page

stapi-fastapi/tests/application.py

@@ -24,10 +24,7 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
     try:
-        yield {
-            "_orders_db": InMemoryOrderDB(),
-            "_opportunities_db": InMemoryOpportunityDB(),
-        }
+        yield {"_orders_db": InMemoryOrderDB(), "_opportunities_db": InMemoryOpportunityDB(), "_opportunities": []}
     finally:
         pass
 

stapi-pydantic/src/stapi_pydantic/__init__.py

@@ -1,6 +1,7 @@
 from .conformance import Conformance
 from .constraints import Constraints
 from .datetime_interval import DatetimeInterval
+from .filter import CQL2Filter
 from .json_schema_model import JsonSchemaModel
 from .opportunity import (
     Opportunity,
@@ -31,6 +32,7 @@
 __all__ = [
     "Conformance",
     "Constraints",
+    "CQL2Filter",
     "DatetimeInterval",
     "JsonSchemaModel",
     "Link",