feat: add stapi-pydantic

.github/workflows/ci.yaml

@@ -0,0 +1,25 @@
+name: "Continuous integration"
+
+concurrency:
+  group: ${{ github.ref }}
+  cancel-in-progress: false
+
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+
+jobs:
+  ci:
+    name: Continuous integration
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v5
+      - name: Sync
+        run: uv sync
+      - name: Lint
+        run: scripts/lint
+      - name: Test
+        run: uv run pytest

.python-version

@@ -0,0 +1 @@
+3.12

pyproject.toml

@@ -0,0 +1,26 @@
+[project]
+name = "pystapi"
+version = "0.0.0" # This package should never be released, only the workspace members should be
+description = "Monorepo for Satellite Tasking API (STAPI) Specification Python packages"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "stapi-pydantic"
+]
+
+[dependency-groups]
+dev = [
+    "mypy>=1.15.0",
+    "pytest>=8.3.5",
+    "ruff>=0.11.2",
+]
+
+[tool.uv.workspace]
+members = ["stapi-pydantic"]
+
+[tool.uv.sources]
+stapi-pydantic.workspace = true
+
+[tool.mypy]
+strict = true
+files = "stapi-pydantic/src/stapi_pydantic/**/*.py"

scripts/format

@@ -0,0 +1,4 @@
+#!/usr/bin/env sh
+
+uv run ruff check --fix
+uv run ruff format

scripts/lint

@@ -0,0 +1,6 @@
+
+#!/usr/bin/env sh
+
+uv run ruff check
+uv run ruff format --check
+uv run mypy

stapi-pydantic/pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "stapi-pydantic"
+version = "0.0.1"
+description = "Pydantic models for Satellite Tasking API (STAPI) Specification"
+readme = "README.md"
+authors = [
+    { name = "Phil Varner", email = "[email protected]" },
+    { name = "Pete Gadomski", email = "[email protected]" },
+]
+requires-python = ">=3.10"
+dependencies = [
+    "cql2>=0.3.6",
+    "geojson-pydantic>=1.2.0",
+]
+
+[project.scripts]
+stapi-pydantic = "stapi_pydantic:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

stapi-pydantic/src/stapi_pydantic/__init__.py

@@ -0,0 +1,11 @@
+from .opportunity import OpportunityProperties
+from .product import Product, Provider, ProviderRole
+from .shared import Link
+
+__all__ = [
+    "Link",
+    "OpportunityProperties",
+    "Product",
+    "Provider",
+    "ProviderRole",
+]

stapi-pydantic/src/stapi_pydantic/conformance.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel, Field
+
+CORE = "https://stapi.example.com/v0.1.0/core"
+OPPORTUNITIES = "https://stapi.example.com/v0.1.0/opportunities"
+ASYNC_OPPORTUNITIES = "https://stapi.example.com/v0.1.0/async-opportunities"
+
+
+class Conformance(BaseModel):
+    conforms_to: list[str] = Field(
+        default_factory=list, serialization_alias="conformsTo"
+    )

stapi-pydantic/src/stapi_pydantic/constraints.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class Constraints(BaseModel):
+    model_config = ConfigDict(extra="allow")

stapi-pydantic/src/stapi_pydantic/datetime_interval.py

@@ -0,0 +1,42 @@
+from datetime import datetime
+from typing import Annotated, Callable
+
+from pydantic import (
+    AfterValidator,
+    AwareDatetime,
+    BeforeValidator,
+    WithJsonSchema,
+    WrapSerializer,
+)
+
+
+def validate_before(
+    value: str | tuple[datetime, datetime],
+) -> tuple[datetime, datetime]:
+    if isinstance(value, str):
+        start, end = value.split("/", 1)
+        return (datetime.fromisoformat(start), datetime.fromisoformat(end))
+    return value
+
+
+def validate_after(value: tuple[datetime, datetime]) -> tuple[datetime, datetime]:
+    if value[1] < value[0]:
+        raise ValueError("end before start")
+    return value
+
+
+def serialize(
+    value: tuple[datetime, datetime],
+    serializer: Callable[[tuple[datetime, datetime]], tuple[str, str]],
+) -> str:
+    del serializer  # unused
+    return f"{value[0].isoformat()}/{value[1].isoformat()}"
+
+
+type DatetimeInterval = Annotated[
+    tuple[AwareDatetime, AwareDatetime],
+    BeforeValidator(validate_before),
+    AfterValidator(validate_after),
+    WrapSerializer(serialize, return_type=str),
+    WithJsonSchema({"type": "string"}, mode="serialization"),
+]

stapi-pydantic/src/stapi_pydantic/filter.py

@@ -0,0 +1,17 @@
+from typing import Annotated, Any
+
+from cql2 import Expr
+from pydantic import BeforeValidator
+
+
+def validate(v: dict[str, Any]) -> dict[str, Any]:
+    if v:
+        expr = Expr(v)
+        expr.validate()
+    return v
+
+
+type CQL2Filter = Annotated[
+    dict,
+    BeforeValidator(validate),
+]

stapi-pydantic/src/stapi_pydantic/json_schema_model.py

@@ -0,0 +1,26 @@
+from typing import Annotated, Any
+
+from pydantic import (
+    BaseModel,
+    PlainSerializer,
+    PlainValidator,
+    WithJsonSchema,
+)
+
+
+def validate(v: Any) -> Any:
+    if not issubclass(v, BaseModel):
+        raise RuntimeError("BaseModel class required")
+    return v
+
+
+def serialize(v: type[BaseModel]) -> dict[str, Any]:
+    return v.model_json_schema()
+
+
+type JsonSchemaModel = Annotated[
+    type[BaseModel],
+    PlainValidator(validate),
+    PlainSerializer(serialize),
+    WithJsonSchema({"type": "object"}),
+]

stapi-pydantic/src/stapi_pydantic/opportunity.py

@@ -0,0 +1,83 @@
+from enum import StrEnum
+from typing import Any, Literal, TypeVar
+
+from geojson_pydantic import Feature, FeatureCollection
+from geojson_pydantic.geometries import Geometry
+from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
+
+from .datetime_interval import DatetimeInterval
+from .filter import CQL2Filter
+from .shared import Link
+
+
+# Copied and modified from https://github.com/stac-utils/stac-pydantic/blob/main/stac_pydantic/item.py#L11
+class OpportunityProperties(BaseModel):
+    datetime: DatetimeInterval
+    product_id: str
+    model_config = ConfigDict(extra="allow")
+
+
+class OpportunityPayload(BaseModel):
+    datetime: DatetimeInterval
+    geometry: Geometry
+    filter: CQL2Filter | None = None
+
+    next: str | None = None
+    limit: int = 10
+
+    model_config = ConfigDict(strict=True)
+
+    def search_body(self) -> dict[str, Any]:
+        return self.model_dump(mode="json", include={"datetime", "geometry", "filter"})
+
+    def body(self) -> dict[str, Any]:
+        return self.model_dump(mode="json")
+
+
+G = TypeVar("G", bound=Geometry)
+P = TypeVar("P", bound=OpportunityProperties)
+
+
+class Opportunity(Feature[G, P]):
+    type: Literal["Feature"] = "Feature"
+    links: list[Link] = Field(default_factory=list)
+
+
+class OpportunityCollection(FeatureCollection[Opportunity[G, P]]):
+    type: Literal["FeatureCollection"] = "FeatureCollection"
+    links: list[Link] = Field(default_factory=list)
+    id: str | None = None
+
+
+class OpportunitySearchStatusCode(StrEnum):
+    received = "received"
+    in_progress = "in_progress"
+    failed = "failed"
+    canceled = "canceled"
+    completed = "completed"
+
+
+class OpportunitySearchStatus(BaseModel):
+    timestamp: AwareDatetime
+    status_code: OpportunitySearchStatusCode
+    reason_code: str | None = None
+    reason_text: str | None = None
+    links: list[Link] = Field(default_factory=list)
+
+
+class OpportunitySearchRecord(BaseModel):
+    id: str
+    product_id: str
+    opportunity_request: OpportunityPayload
+    status: OpportunitySearchStatus
+    links: list[Link] = Field(default_factory=list)
+
+
+class OpportunitySearchRecords(BaseModel):
+    search_records: list[OpportunitySearchRecord]
+    links: list[Link] = Field(default_factory=list)
+
+
+class Prefer(StrEnum):
+    respond_async = "respond-async"
+    wait = "wait"