feat: stac-geoparquet backend (#29)

- Closes #28

Author: gadomski

data/naip.parquet

@@ -7,6 +7,8 @@ requires-python = ">=3.12"
 dependencies = [
     "fastapi[standard]>=0.115.6",
     "pydantic>=2.10.4",
+    "pydantic-settings>=2.7.1",
+    "stacrs>=0.3.0",
 ]
 
 [dependency-groups]
@@ -18,6 +20,7 @@ files = ["**/*.py"]
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
+asyncio_default_fixture_loop_scope = "function"
 
 [tool.ruff.lint]
 select = ["E", "F", "I"]

pyproject.toml

@@ -2,4 +2,4 @@
 
 set -e
 
-uv run fastapi dev
+TISTAC_BACKEND=data/naip.parquet uv run fastapi dev

scripts/dev

@@ -0,0 +1,3 @@
+from .settings import Settings
+
+__all__ = ["Settings"]

src/tistac/__init__.py

@@ -1,12 +1,28 @@
-from fastapi import FastAPI
+from typing import Annotated
 
+from fastapi import Depends, FastAPI, Query
+
+from .backend import Backend
 from .item_collection import ItemCollection
 from .root import Root
+from .search import GetSearch, Search
+from .settings import Settings
 
 
-def build() -> FastAPI:
+def build(settings: Settings | None = None) -> FastAPI:
     """Builds a new TiStac application."""
 
+    if settings is None:
+        settings = Settings()  # type: ignore
+
+    async def get_settings() -> Settings:
+        return settings
+
+    backend = settings.get_backend()
+
+    async def get_backend() -> Backend:
+        return backend
+
     app = FastAPI()
 
     @app.get("/")
@@ -16,15 +32,24 @@ async def root() -> Root:
         return Root()
 
     @app.get("/search")
-    async def get_search() -> ItemCollection:
+    async def get_search(
+        get_search: Annotated[GetSearch, Query()],
+        backend: Annotated[Backend, Depends(get_backend)],
+        settings: Annotated[Settings, Depends(get_settings)],
+    ) -> ItemCollection:
         """Searches this STAC API via a GET request."""
-
-        return ItemCollection()
+        search = get_search.into_search()
+        search = settings.update_search(search)
+        return await backend.search(search)
 
     @app.post("/search")
-    async def post_search() -> ItemCollection:
+    async def post_search(
+        search: Search,
+        backend: Annotated[Backend, Depends(get_backend)],
+        settings: Annotated[Settings, Depends(get_settings)],
+    ) -> ItemCollection:
         """Searches this STAC API via a POST request."""
-
-        return ItemCollection()
+        search = settings.update_search(search)
+        return await backend.search(search)
 
     return app

src/tistac/app.py

@@ -0,0 +1,12 @@
+from abc import ABC, abstractmethod
+
+from .item_collection import ItemCollection
+from .search import Search
+
+
+class Backend(ABC):
+    """A TiStac backend."""
+
+    @abstractmethod
+    async def search(self, search: Search) -> ItemCollection:
+        """Searches this backend."""

src/tistac/backend.py

@@ -0,0 +1,2 @@
+DEFAULT_LIMIT = 20
+"""The default page size."""

src/tistac/constants.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import BaseModel
 
 
@@ -7,3 +9,5 @@ class ItemCollection(BaseModel):
     The features of this collection may not be valid STAC items, if the `fields`
     extension is used.
     """
+
+    features: list[dict[str, Any]]

src/tistac/item_collection.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel, Field
+
+
+class Search(BaseModel):
+    """A STAC API POST search."""
+
+    limit: int | None = Field(default=None)
+    """The maximum number of items per page."""
+
+
+class GetSearch(BaseModel):
+    """A STAC API GET search."""
+
+    def into_search(self) -> Search:
+        """Converts this GET search into a POST search."""
+        return Search()

src/tistac/search.py

@@ -0,0 +1,27 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from .backend import Backend
+from .constants import DEFAULT_LIMIT
+from .search import Search
+
+
+class Settings(BaseSettings):
+    """TiStac settings."""
+
+    model_config = SettingsConfigDict(env_prefix="tistac_")
+
+    backend: str
+    default_limit: int = Field(default=DEFAULT_LIMIT)
+
+    def get_backend(self) -> Backend:
+        """Returns the configured backend."""
+        from .stac_geoparquet import StacGeoparquetBackend
+
+        return StacGeoparquetBackend(self.backend)
+
+    def update_search(self, search: Search) -> Search:
+        """Updates a search with some default settings."""
+        if search.limit is None:
+            search.limit = self.default_limit
+        return search