Pydantic validation of extensions

CHANGES.md

@@ -41,6 +41,10 @@
 
 ### Added
 
+- add `validate_extensions` setting that enables validation of `stac_extensions` from submitted STAC objects
+  using the `stac_pydantic.extensions.validate_extensions` utility. Applicable only when `TransactionExtension`
+  is active.
+- add `validation` extra requirement to install dependencies of `stac_pydantic` required for extension validation
 - add `write_connection_pool` option in `stac_fastapi.pgstac.db.connect_to_db` function
 - add `write_postgres_settings` option in `stac_fastapi.pgstac.db.connect_to_db` function to set specific settings for the `writer` DB connection pool
 - add specific error message when trying to create `Item` with null geometry (not supported by PgSTAC)

setup.py

@@ -45,6 +45,9 @@
     ],
     "server": ["uvicorn[standard]==0.35.0"],
     "awslambda": ["mangum"],
+    "validation": [
+        "stac_pydantic[validation]",
+    ],
 }
 
 

stac_fastapi/pgstac/config.py

@@ -168,6 +168,13 @@ class Settings(ApiSettings):
     invalid_id_chars: List[str] = DEFAULT_INVALID_ID_CHARS
     base_item_cache: Type[BaseItemCache] = DefaultBaseItemCache
 
+    validate_extensions: bool = False
+    """
+    Validate `stac_extensions` schemas against submitted data when creating or updated STAC objects.
+
+    Implies that the `Transactions` extension is enabled.
+    """
+
     cors_origins: str = "*"
     cors_methods: str = "GET,POST,OPTIONS"
 

stac_fastapi/pgstac/transactions.py

@@ -2,7 +2,7 @@
 
 import logging
 import re
-from typing import List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 import attr
 from buildpg import render
@@ -20,6 +20,7 @@
 )
 from stac_fastapi.types import stac as stac_types
 from stac_pydantic import Collection, Item, ItemCollection
+from stac_pydantic.extensions import validate_extensions
 from starlette.responses import JSONResponse, Response
 
 from stac_fastapi.pgstac.config import Settings
@@ -41,8 +42,35 @@ def _validate_id(self, id: str, settings: Settings):
                 detail=f"ID ({id}) cannot contain the following characters: {' '.join(invalid_chars)}",
             )
 
+    def _validate_extensions(
+        self,
+        stac_object: stac_types.Item | stac_types.Collection | stac_types.Catalog | Dict[str, Any],
+        settings: Settings,
+    ) -> None:
+        """Validate extensions of the STAC object data."""
+        if not settings.validate_extensions:
+            return
+        if isinstance(stac_object, dict):
+            if not stac_object.get("stac_extensions"):
+                return
+        else:
+            if not stac_object.stac_extensions:
+                return
+
+        try:
+            validate_extensions(
+                stac_object,
+                reraise_exception=True,
+            )
+        except Exception as err:
+            raise HTTPException(
+                status_code=422,
+                detail=f"STAC Extensions failed validation: {err!s}",
+            ) from err
+
     def _validate_collection(self, request: Request, collection: stac_types.Collection):
         self._validate_id(collection["id"], request.app.state.settings)
+        self._validate_extensions(collection, request.app.state.settings)
 
     def _validate_item(
         self,
@@ -56,6 +84,7 @@ def _validate_item(
         body_item_id = item.get("id")
 
         self._validate_id(body_item_id, request.app.state.settings)
+        self._validate_extensions(item, request.app.state.settings)
 
         if item.get("geometry", None) is None:
             raise HTTPException(
@@ -177,6 +206,7 @@ async def update_collection(
         """Update collection."""
 
         col = collection.model_dump(mode="json")
+        self._validate_collection(request, col)
 
         async with request.app.state.get_connection(request, "w") as conn:
             await dbfunc(conn, "update_collection", col)