new structure

Author: nic-planet

pyproject.toml

@@ -1,9 +1,9 @@
 [tool.poetry]
-name = "stapi-fastapi"
+name = "stapi-fastapi-planet"
 # placeholder version filled by poetry-dynamic-versioning
 version = "0.0.0"
-description = "Sensor Tasking API (STAPI) with FastAPI"
-authors = ["Christian Wygoda <[email protected]>"]
+description = "Sensor Tasking API (STAPI) with FastAPI - Planet proxy"
+authors = ["Nicolas Neubauer <[email protected]>", "Christian Wygoda <[email protected]>"]
 license = "MIT"
 readme = "README.md"
 packages = [{include = "stapi_fastapi", from="src"}]

src/planet/README.md

@@ -0,0 +1,21 @@
+# STAT FastAPI - Planet
+
+This is an example implementation for `https://github.com/stat-utils/stat-fastapi` proxying to the Planet Tasking API.
+
+
+Start the server locally
+
+```sh
+uvicorn planet.application:app --app-dir ./src/planet --reload
+```
+
+GET all products
+```sh
+curl http://127.0.0.1:8000/products
+```
+
+POST to opportunities
+```sh
+export BACKEND_TOKEN=...
+curl -d '{"geometry": {"type": "Point", "coordinates": [13.4, 52.5]}, "product_id": "PL-123456:Assured Tasking", "datetime": "2024-05-01T00:00:00Z/2024-05-12T00:00:00Z"}' -H "Content-Type: application/json; Authorization: $BACKEND_TOKEN" -X POST http://127.0.0.1:8000/opportunities
+```

src/planet/__init__.py

@@ -0,0 +1,51 @@
+import os
+import sys
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from typing import Any
+
+from fastapi import FastAPI
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from stapi_fastapi.models.conformance import ASYNC_OPPORTUNITIES, CORE, OPPORTUNITIES
+from stapi_fastapi.routers.root_router import RootRouter
+
+from backends import (
+    mock_get_opportunity_search_record,
+    mock_get_opportunity_search_records,
+    get_order,
+    mock_get_order_statuses,
+    mock_get_orders,
+)
+from shared import (
+    InMemoryOpportunityDB,
+    InMemoryOrderDB,
+    product_test_planet_sync_opportunity
+    #product_test_spotlight_sync_async_opportunity,
+)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI) -> AsyncIterator[dict[str, Any]]:
+    try:
+        yield {
+            "_orders_db": InMemoryOrderDB(),
+            "_opportunities_db": InMemoryOpportunityDB(),
+        }
+    finally:
+        pass
+
+
+root_router = RootRouter(
+    get_orders=mock_get_orders,
+    get_order=get_order,
+    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], # , ASYNC_OPPORTUNITIES
+)
+#root_router.add_product(product_test_spotlight_sync_async_opportunity)
+root_router.add_product(product_test_planet_sync_opportunity)
+app: FastAPI = FastAPI(lifespan=lifespan)
+app.include_router(root_router, prefix="")

src/planet/application.py

@@ -0,0 +1,227 @@
+from datetime import datetime, timezone
+from uuid import uuid4
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+from fastapi import Request
+from returns.maybe import Maybe, Nothing, Some
+from returns.result import Failure, ResultE, Success
+
+from stapi_fastapi.models.opportunity import (
+    Opportunity,
+    OpportunityCollection,
+    OpportunityPayload,
+    OpportunitySearchRecord,
+    OpportunitySearchStatus,
+    OpportunitySearchStatusCode,
+)
+from stapi_fastapi.models.order import (
+    Order,
+    OrderPayload,
+    OrderProperties,
+    OrderSearchParameters,
+    OrderStatus,
+    OrderStatusCode,
+)
+from stapi_fastapi.routers.product_router import ProductRouter
+
+from client import Client
+from conversions import planet_order_to_stapi_order, stapi_opportunity_payload_to_planet_iw_search
+
+
+async def mock_get_orders(
+    next: str | None, limit: int, request: Request
+) -> ResultE[tuple[list[Order], Maybe[str]]]:
+    """
+    Return orders from backend.  Handle pagination/limit if applicable
+    """
+    try:
+        start = 0
+        limit = min(limit, 100)
+
+        order_ids = [*request.state._orders_db._orders.keys()]
+
+        if next:
+            start = order_ids.index(next)
+        end = start + limit
+        ids = order_ids[start:end]
+        orders = [request.state._orders_db.get_order(order_id) for order_id in ids]
+
+        if end > 0 and end < len(order_ids):
+            return Success(
+                (orders, Some(request.state._orders_db._orders[order_ids[end]].id))
+            )
+        return Success((orders, Nothing))
+    except Exception as e:
+        return Failure(e)
+
+
+async def get_order(order_id: str, request: Request) -> ResultE[Maybe[Order]]:
+    """
+    Show details for order with `order_id`.
+    """
+    try:
+        return Success(
+            Maybe.from_optional(planet_order_to_stapi_order(Client(request).get_order(order_id)))
+        )
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_get_order_statuses(
+    order_id: str, next: str | None, limit: int, request: Request
+) -> ResultE[Maybe[tuple[list[OrderStatus], Maybe[str]]]]:
+    try:
+        start = 0
+        limit = min(limit, 100)
+        statuses = request.state._orders_db.get_order_statuses(order_id)
+        if statuses is None:
+            return Success(Nothing)
+
+        if next:
+            start = int(next)
+        end = start + limit
+        stati = statuses[start:end]
+
+        if end > 0 and end < len(statuses):
+            return Success(Some((stati, Some(str(end)))))
+        return Success(Some((stati, Nothing)))
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_create_order(
+    product_router: ProductRouter, payload: OrderPayload, request: Request
+) -> ResultE[Order]:
+    """
+    Create a new order.
+    """
+    try:
+        status = OrderStatus(
+            timestamp=datetime.now(timezone.utc),
+            status_code=OrderStatusCode.received,
+        )
+        order = Order(
+            id=str(uuid4()),
+            geometry=payload.geometry,
+            properties=OrderProperties(
+                product_id=product_router.product.id,
+                created=datetime.now(timezone.utc),
+                status=status,
+                search_parameters=OrderSearchParameters(
+                    geometry=payload.geometry,
+                    datetime=payload.datetime,
+                    filter=payload.filter,
+                ),
+                order_parameters=payload.order_parameters.model_dump(),
+                opportunity_properties={
+                    "datetime": "2024-01-29T12:00:00Z/2024-01-30T12:00:00Z",
+                    "off_nadir": 10,
+                },
+            ),
+            links=[],
+        )
+
+        request.state._orders_db.put_order(order)
+        request.state._orders_db.put_order_status(order.id, status)
+        return Success(order)
+    except Exception as e:
+        return Failure(e)
+
+
+# TODO why does this return a list of Opportunities and not an OpportunityCollection?
+async def search_opportunities(
+    product_router: ProductRouter,
+    search: OpportunityPayload,
+    next: str | None,
+    limit: int,
+    request: Request,
+) -> ResultE[tuple[list[Opportunity], Maybe[str]]]:
+    try:
+        iw_request = stapi_opportunity_payload_to_planet_iw_search(product_router.product, search)
+        imaging_windows = Client(request).get_imaging_windows(iw_request)
+
+        opportunities = [
+            imaging_window_to_opportunity(iw, search)
+            for iw
+            in imaging_windows
+        ]
+        #return OpportunityCollection(features=opportunities)
+        return Success((opportunities, Nothing))
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_search_opportunities_async(
+    product_router: ProductRouter,
+    search: OpportunityPayload,
+    request: Request,
+) -> ResultE[OpportunitySearchRecord]:
+    try:
+        received_status = OpportunitySearchStatus(
+            timestamp=datetime.now(timezone.utc),
+            status_code=OpportunitySearchStatusCode.received,
+        )
+        search_record = OpportunitySearchRecord(
+            id=str(uuid4()),
+            product_id=product_router.product.id,
+            opportunity_request=search,
+            status=received_status,
+            links=[],
+        )
+        request.state._opportunities_db.put_search_record(search_record)
+        return Success(search_record)
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_get_opportunity_collection(
+    product_router: ProductRouter, opportunity_collection_id: str, request: Request
+) -> ResultE[Maybe[OpportunityCollection]]:
+    try:
+        return Success(
+            Maybe.from_optional(
+                request.state._opportunities_db.get_opportunity_collection(
+                    opportunity_collection_id
+                )
+            )
+        )
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_get_opportunity_search_records(
+    next: str | None,
+    limit: int,
+    request: Request,
+) -> ResultE[tuple[list[OpportunitySearchRecord], Maybe[str]]]:
+    try:
+        start = 0
+        limit = min(limit, 100)
+        search_records = request.state._opportunities_db.get_search_records()
+
+        if next:
+            start = int(next)
+        end = start + limit
+        page = search_records[start:end]
+
+        if end > 0 and end < len(search_records):
+            return Success((page, Some(str(end))))
+        return Success((page, Nothing))
+    except Exception as e:
+        return Failure(e)
+
+
+async def mock_get_opportunity_search_record(
+    search_record_id: str, request: Request
+) -> ResultE[Maybe[OpportunitySearchRecord]]:
+    try:
+        return Success(
+            Maybe.from_optional(
+                request.state._opportunities_db.get_search_record(search_record_id)
+            )
+        )
+    except Exception as e:
+        return Failure(e)

src/planet/backends.py

@@ -0,0 +1,52 @@
+import os
+import time
+
+import requests
+from fastapi import Request
+
+from settings import Settings
+
+class Client:
+
+    def __init__(self, request: Request):
+
+        token = os.environ.get("BACKEND_TOKEN")
+        if authorization := request.headers.get("authorization"):
+            token = authorization.replace("Bearer ", "")
+
+        self.token = token
+        self.headers = {
+            'Content-Type': 'application/json',
+            'Authorization': f'api-key {self.token}'
+        }
+        self.orders_url = f'{Settings().api_base_url}/orders/'
+        self.iw_search_url = f'{Settings().api_base_url}/imaging-windows/search'
+
+    def get_order(self, order_id: str) -> dict:
+        order_url = f'{self.orders_url}{order_id}'
+        response = requests.get(order_url, headers=self.headers, allow_redirects=False)
+        response.raise_for_status()
+        return response.json()
+
+    # todo this is a sync wrapper around an async search, migrate to async
+    def get_imaging_windows(self, payload: dict) -> dict:
+        iw_url = f'{self.iw_search_url}'
+        r = requests.post(iw_url, json=payload, headers=self.headers, allow_redirects=False)
+        r.raise_for_status()
+        if 'location' not in r.headers:
+            raise ValueError(
+                "Header 'location' not found: %s, status %s, body %s" % (
+                    list(r.headers.keys()), r.status_code, r.text)
+            )
+        poll_url = r.headers['location']
+
+        while True:
+            r = requests.get(poll_url, headers=self.headers)
+            r.raise_for_status()
+            status = r.json()['status']
+            if status == "DONE":
+                return r.json()['imaging_windows']
+            elif status == 'FAILED':
+                raise ValueError(
+                    f"Retrieving Imaging Windows failed: {r.json['error_code']} - {r.json['error_message']}'")
+            time.sleep(1)