第08章 - stac-fastapi 服务开发

def crop(v: PositiveInt) -> PositiveInt:
    limit = 10_000
    if v > limit:
        v = limit
    return v

Limit = Annotated[PositiveInt, AfterValidator(crop)]

from fastapi import FastAPI
from stac_fastapi.api.app import StacApi
from stac_fastapi.types.config import ApiSettings

from mybackend import MemoryClient

settings = ApiSettings(
    stac_fastapi_title="我的影像目录",
    stac_fastapi_description="示例 STAC API",
    stac_fastapi_version="1.0.0",
    stac_fastapi_landing_id="my-catalog",
)

api = StacApi(
    settings=settings,
    client=MemoryClient(),
    extensions=[],
)

app: FastAPI = api.app

# 1. settings 注入 client；2. 设置全局 settings 单例
self.client.title = self.title
self.client.description = self.description
Settings.set(self.settings)

# 3. 注册核心路由；4. 逐个注册扩展
self.register_core()
for ext in self.extensions:
    ext.register(self.app)

from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

middlewares = [
    Middleware(
        CORSMiddleware,
        allow_origins=["https://app.example.com"],
        allow_methods=["GET", "POST", "OPTIONS"],
        allow_headers=["Content-Type", "Authorization", "Range"],
        expose_headers=["Content-Range", "Accept-Ranges"],
        max_age=600,
    ),
]

# mybackend.py
from typing import Any

import attr
from stac_fastapi.types.core import AsyncBaseCoreClient
from stac_fastapi.types.stac import Item, ItemCollection, Collection, Collections
from stac_fastapi.types.search import BaseSearchPostRequest


def _bbox_hit(item_bbox, query_bbox) -> bool:
    if not item_bbox or not query_bbox:
        return True
    if len(query_bbox) == 4:
        w, s, e, n = query_bbox
    else:
        w, s, _, e, n, _ = query_bbox
    return not (item_bbox[2] < w or item_bbox[0] > e
                or item_bbox[3] < s or item_bbox[1] > n)


@attr.s
class MemoryClient(AsyncBaseCoreClient):
    """极简内存后端，仅用于演示与测试。"""

    collections: dict[str, dict] = attr.ib(factory=dict)
    items: dict[tuple[str, str], dict] = attr.ib(factory=dict)

    def seed(self, collection: dict, items: list[dict]) -> None:
        cid = collection["id"]
        self.collections[cid] = collection
        for it in items:
            self.items[(cid, it["id"])] = it

    async def all_collections(self, **kwargs) -> Collections:
        return Collections(
            collections=list(self.collections.values()),
            links=[],
            numberMatched=len(self.collections),
            numberReturned=len(self.collections),
        )

    async def get_collection(self, collection_id: str, **kwargs) -> Collection:
        if collection_id not in self.collections:
            raise NotFoundError(f"collection {collection_id} not found")
        return Collection(**self.collections[collection_id])

    async def get_item(self, item_id: str, collection_id: str, **kwargs) -> Item:
        key = (collection_id, item_id)
        if key not in self.items:
            raise NotFoundError(f"item {item_id} not found")
        return Item(**self.items[key])

    async def item_collection(
        self, collection_id: str, bbox=None, datetime=None,
        limit: int = 10, token: str | None = None, **kwargs,
    ) -> ItemCollection:
        rows = [v for (cid, _), v in self.items.items() if cid == collection_id]
        rows = [r for r in rows if _bbox_hit(r.get("bbox"), bbox)]
        return ItemCollection(
            type="FeatureCollection",
            features=rows[:limit],
            links=[],
            numberReturned=len(rows[:limit]),
            numberMatched=len(rows),
        )

    async def get_search(
        self, collections=None, ids=None, bbox=None, intersects=None,
        datetime=None, limit: int | None = 10, **kwargs,
    ) -> ItemCollection:
        rows = list(self.items.values())
        if collections:
            rows = [r for r in rows if r.get("collection") in collections]
        if ids:
            rows = [r for r in rows if r["id"] in ids]
        rows = [r for r in rows if _bbox_hit(r.get("bbox"), bbox)]
        n = limit or 10
        return ItemCollection(
            type="FeatureCollection",
            features=rows[:n],
            links=[],
            numberReturned=len(rows[:n]),
            numberMatched=len(rows),
        )

    async def post_search(
        self, search_request: BaseSearchPostRequest, **kwargs
    ) -> ItemCollection:
        return await self.get_search(
            collections=search_request.collections,
            ids=search_request.ids,
            bbox=search_request.bbox,
            intersects=search_request.intersects,
            datetime=search_request.datetime,
            limit=search_request.limit,
        )

# main.py
from fastapi import FastAPI
from stac_fastapi.api.app import StacApi
from stac_fastapi.types.config import ApiSettings

from mybackend import MemoryClient

settings = ApiSettings(stac_fastapi_title="内存示例")
client = MemoryClient()
client.seed(
    collection={
        "type": "Collection", "stac_version": "1.1.0", "id": "demo",
        "description": "示例集合", "license": "CC-BY-4.0",
        "extent": {"spatial": {"bbox": [[100, 20, 110, 30]]},
                   "temporal": {"interval": [["2023-01-01T00:00:00Z", None]]}},
        "links": [],
    },
    items=[{
        "type": "Feature", "stac_version": "1.1.0", "id": "demo-001", "collection": "demo",
        "geometry": {"type": "Polygon", "coordinates": [
            [[100, 20], [110, 20], [110, 30], [100, 30], [100, 20]]]},
        "bbox": [100, 20, 110, 30],
        "properties": {"datetime": "2023-05-01T00:00:00Z"}, "links": [], "assets": {},
    }],
)

api = StacApi(settings=settings, client=client, extensions=[])
app: FastAPI = api.app

class TransactionConformanceClasses(StrEnum):
    ITEMS = "https://api.stacspec.org/v1.0.0/ogcapi-features/extensions/transaction"
    COLLECTIONS = "https://api.stacspec.org/v1.0.0/collections/extensions/transaction"

from stac_fastapi.extensions import TransactionExtension
from stac_fastapi.api.models import JSONResponse

from mybackend import MemoryTransactionsClient

api = StacApi(
    settings=settings,
    client=client,
    extensions=[
        TransactionExtension(
            client=MemoryTransactionsClient(store=client),
            settings=settings,
            response_class=JSONResponse,
        )
    ],
)

def create_item(self, collection_id: str, item: Item, **kwargs): ...
def update_item(self, collection_id: str, item_id: str, item: Item, **kwargs): ...
def patch_item(self, collection_id: str, item_id: str, patch, **kwargs): ...
def delete_item(self, item_id: str, collection_id: str, **kwargs): ...   # 注意顺序
def create_collection(self, collection: Collection, **kwargs): ...
def update_collection(self, collection_id: str, collection: Collection, **kwargs): ...
def patch_collection(self, collection_id: str, patch, **kwargs): ...
def delete_collection(self, collection_id: str, **kwargs): ...

async def create_item(request, response, request_data=Depends(PostItem)):
    item = await endpoint(request, request_data)
    # 单条 Item 成功创建时补 Location；ItemCollection 不补
    if request_data.item.type == "Feature" and "location" not in response.headers:
        response.headers["Location"] = _location_url(
            request, "Get Item", collection_id=..., item_id=item["id"])
    return item

import attr
from fastapi import FastAPI
from stac_fastapi.types.extension import ApiExtension

@attr.s
class StatsExtension(ApiExtension):
    client: object = attr.ib(default=None)

    conformance_classes: list[str] = attr.ib(default=[])
    schema_href: str | None = attr.ib(default=None)

    def register(self, app: FastAPI) -> None:
        @app.get("/stats")
        async def stats():
            return {"collections": len(self.client.collections)}

api = StacApi(settings=settings, client=client, extensions=[StatsExtension(client=client)])

def create_app() -> FastAPI:
    api = StacApi(settings=Settings(), client=CoreClient(), extensions=[...])
    return api.app