PgSTAC: API hydration of search result items

CHANGES.md

@@ -17,6 +17,7 @@
 * Bulk Transactions object Items iterator now returns the Item objects rather than the string IDs of the Item objects
   ([#355](https://github.com/stac-utils/stac-fastapi/issues/355))
 * docker-compose now runs uvicorn with hot-reloading enabled
+* Bump version of PGStac to 0.6.2 that includes support for hydrating results in the API backed ([#397](https://github.com/stac-utils/stac-fastapi/pull/397))
 
 ### Removed
 
@@ -27,7 +28,8 @@
 * Fixes issues (and adds tests) for issues caused by regression in pgstac ([#345](https://github.com/stac-utils/stac-fastapi/issues/345)
 * Update error response payloads to match the API spec. ([#361](https://github.com/stac-utils/stac-fastapi/pull/361))
 * Fixed stray `/` before the `#` in several extension conformance class strings ([383](https://github.com/stac-utils/stac-fastapi/pull/383))
-* SQLAlchemy backend bulk item insert now works ([#356]https://github.com/stac-utils/stac-fastapi/issues/356))
+* SQLAlchemy backend bulk item insert now works ([#356](https://github.com/stac-utils/stac-fastapi/issues/356))
+* PGStac Backend has stricter implementation of Fields Extension syntax ([#397](https://github.com/stac-utils/stac-fastapi/pull/397))
 
 ## [2.3.0]
 

docker-compose.yml

@@ -50,6 +50,7 @@ services:
       - GDAL_DISABLE_READDIR_ON_OPEN=EMPTY_DIR
       - DB_MIN_CONN_SIZE=1
       - DB_MAX_CONN_SIZE=1
+      - USE_API_HYDRATE=${USE_API_HYDRATE:-false}
     ports:
       - "8082:8082"
     volumes:
@@ -62,7 +63,7 @@ services:
 
   database:
     container_name: stac-db
-    image: ghcr.io/stac-utils/pgstac:v0.4.5
+    image: ghcr.io/stac-utils/pgstac:v0.6.2
     environment:
       - POSTGRES_USER=username
       - POSTGRES_PASSWORD=password

stac_fastapi/api/stac_fastapi/api/errors.py

@@ -23,7 +23,7 @@
 DEFAULT_STATUS_CODES = {
     NotFoundError: status.HTTP_404_NOT_FOUND,
     ConflictError: status.HTTP_409_CONFLICT,
-    ForeignKeyError: status.HTTP_422_UNPROCESSABLE_ENTITY,
+    ForeignKeyError: status.HTTP_424_FAILED_DEPENDENCY,
     DatabaseError: status.HTTP_424_FAILED_DEPENDENCY,
     Exception: status.HTTP_500_INTERNAL_SERVER_ERROR,
     InvalidQueryParameter: status.HTTP_400_BAD_REQUEST,

stac_fastapi/pgstac/README.md

@@ -46,7 +46,12 @@ pip install -e \
     stac_fastapi/pgstac[dev,server]
 ```
 
+### Settings
+
+To configure PGStac stac-fastapi to [hydrate search result items in the API](https://github.com/stac-utils/pgstac#runtime-configurations), set the `USE_API_HYDRATE` environment variable to `true` or explicitly set the option in the PGStac Settings object.
+
 ### Migrations
+
 PGStac is an external project and the may be used by multiple front ends.
 For Stac FastAPI development, a docker image (which is pulled as part of the docker-compose) is available at
 bitner/pgstac:[version] that has the full database already set up for PGStac.

stac_fastapi/pgstac/setup.py

@@ -17,16 +17,17 @@
     "buildpg",
     "brotli_asgi",
     "pygeofilter @ git+https://github.com/geopython/[email protected]#egg=pygeofilter",
+    "pypgstac==0.6.*",
 ]
 
 extra_reqs = {
     "dev": [
+        "pypgstac[psycopg]==0.6.*",
         "pytest",
         "pytest-cov",
         "pytest-asyncio>=0.17",
         "pre-commit",
         "requests",
-        "pypgstac==0.4.5",
         "httpx",
     ],
     "docs": ["mkdocs", "mkdocs-material", "pdocs"],

stac_fastapi/pgstac/stac_fastapi/pgstac/config.py

@@ -1,5 +1,11 @@
 """Postgres API configuration."""
 
+from typing import Type
+
+from stac_fastapi.pgstac.types.base_item_cache import (
+    BaseItemCache,
+    DefaultBaseItemCache,
+)
 from stac_fastapi.types.config import ApiSettings
 
 
@@ -13,6 +19,7 @@ class Settings(ApiSettings):
         postgres_host_writer: hostname for the writer connection.
         postgres_port: database port.
         postgres_dbname: database name.
+        use_api_hydrate: perform hydration of stac items within stac-fastapi.
     """
 
     postgres_user: str
@@ -27,6 +34,9 @@ class Settings(ApiSettings):
     db_max_queries: int = 50000
     db_max_inactive_conn_lifetime: float = 300
 
+    use_api_hydrate: bool = False
+    base_item_cache: Type[BaseItemCache] = DefaultBaseItemCache
+
     testing: bool = False
 
     @property

stac_fastapi/pgstac/stac_fastapi/pgstac/core.py

@@ -12,12 +12,15 @@
 from pydantic import ValidationError
 from pygeofilter.backends.cql2_json import to_cql2
 from pygeofilter.parsers.cql2_text import parse as parse_cql2_text
+from pypgstac.hydration import hydrate
 from stac_pydantic.links import Relations
 from stac_pydantic.shared import MimeTypes
 from starlette.requests import Request
 
+from stac_fastapi.pgstac.config import Settings
 from stac_fastapi.pgstac.models.links import CollectionLinks, ItemLinks, PagingLinks
 from stac_fastapi.pgstac.types.search import PgstacSearch
+from stac_fastapi.pgstac.utils import filter_fields
 from stac_fastapi.types.core import AsyncBaseCoreClient
 from stac_fastapi.types.errors import InvalidQueryParameter, NotFoundError
 from stac_fastapi.types.stac import Collection, Collections, Item, ItemCollection
@@ -103,8 +106,38 @@ async def get_collection(self, collection_id: str, **kwargs) -> Collection:
 
         return Collection(**collection)
 
+    async def _get_base_item(
+        self, collection_id: str, request: Request
+    ) -> Dict[str, Any]:
+        """Get the base item of a collection for use in rehydrating full item collection properties.
+
+        Args:
+            collection: ID of the collection.
+
+        Returns:
+            Item.
+        """
+        item: Optional[Dict[str, Any]]
+
+        pool = request.app.state.readpool
+        async with pool.acquire() as conn:
+            q, p = render(
+                """
+                SELECT * FROM collection_base_item(:collection_id::text);
+                """,
+                collection_id=collection_id,
+            )
+            item = await conn.fetchval(q, *p)
+
+        if item is None:
+            raise NotFoundError(f"A base item for {collection_id} does not exist.")
+
+        return item
+
     async def _search_base(
-        self, search_request: PgstacSearch, **kwargs: Any
+        self,
+        search_request: PgstacSearch,
+        **kwargs: Any,
     ) -> ItemCollection:
         """Cross catalog search (POST).
 
@@ -119,9 +152,11 @@ async def _search_base(
         items: Dict[str, Any]
 
         request: Request = kwargs["request"]
+        settings: Settings = request.app.state.settings
         pool = request.app.state.readpool
 
-        # pool = kwargs["request"].app.state.readpool
+        search_request.conf = search_request.conf or {}
+        search_request.conf["nohydrate"] = settings.use_api_hydrate
         req = search_request.json(exclude_none=True, by_alias=True)
 
         try:
@@ -141,30 +176,65 @@ async def _search_base(
         next: Optional[str] = items.pop("next", None)
         prev: Optional[str] = items.pop("prev", None)
         collection = ItemCollection(**items)
-        cleaned_features: List[Item] = []
 
-        for feature in collection.get("features") or []:
-            feature = Item(**feature)
+        exclude = search_request.fields.exclude
+        if exclude and len(exclude) == 0:
+            exclude = None
+        include = search_request.fields.include
+        if include and len(include) == 0:
+            include = None
+
+        async def _add_item_links(
+            feature: Item,
+            collection_id: Optional[str] = None,
+            item_id: Optional[str] = None,
+        ) -> None:
+            """Add ItemLinks to the Item.
+
+            If the fields extension is excluding links, then don't add them.
+            Also skip links if the item doesn't provide collection and item ids.
+            """
+            collection_id = feature.get("collection") or collection_id
+            item_id = feature.get("id") or item_id
+
             if (
                 search_request.fields.exclude is None
                 or "links" not in search_request.fields.exclude
+                and all([collection_id, item_id])
             ):
-                # TODO: feature.collection is not always included
-                # This code fails if it's left outside of the fields expression
-                # I've fields extension updated test cases to always include feature.collection
                 feature["links"] = await ItemLinks(
-                    collection_id=feature["collection"],
-                    item_id=feature["id"],
+                    collection_id=collection_id,
+                    item_id=item_id,
                     request=request,
                 ).get_links(extra_links=feature.get("links"))
 
-                exclude = search_request.fields.exclude
-                if exclude and len(exclude) == 0:
-                    exclude = None
-                include = search_request.fields.include
-                if include and len(include) == 0:
-                    include = None
-            cleaned_features.append(feature)
+        cleaned_features: List[Item] = []
+
+        if settings.use_api_hydrate:
+
+            async def _get_base_item(collection_id: str) -> Dict[str, Any]:
+                return await self._get_base_item(collection_id, request)
+
+            base_item_cache = settings.base_item_cache(
+                fetch_base_item=_get_base_item, request=request
+            )
+
+            for feature in collection.get("features") or []:
+                base_item = await base_item_cache.get(feature.get("collection"))
+                feature = hydrate(base_item, feature)
+
+                # Grab ids needed for links that may be removed by the fields extension.
+                collection_id = feature.get("collection")
+                item_id = feature.get("id")
+
+                feature = filter_fields(feature, include, exclude)
+                await _add_item_links(feature, collection_id, item_id)
+
+                cleaned_features.append(feature)
+        else:
+            for feature in collection.get("features") or []:
+                await _add_item_links(feature)
+                cleaned_features.append(feature)
 
         collection["features"] = cleaned_features
         collection["links"] = await PagingLinks(

stac_fastapi/pgstac/stac_fastapi/pgstac/types/base_item_cache.py

@@ -0,0 +1,55 @@
+"""base_item_cache classes for pgstac fastapi."""
+import abc
+from typing import Any, Callable, Coroutine, Dict
+
+from starlette.requests import Request
+
+
+class BaseItemCache(abc.ABC):
+    """
+    A cache that returns a base item for a collection.
+
+    If no base item is found in the cache, use the fetch_base_item function
+    to fetch the base item from pgstac.
+    """
+
+    def __init__(
+        self,
+        fetch_base_item: Callable[[str], Coroutine[Any, Any, Dict[str, Any]]],
+        request: Request,
+    ):
+        """
+        Initialize the base item cache.
+
+        Args:
+            fetch_base_item: A function that fetches the base item for a collection.
+            request: The request object containing app state that may be used by caches.
+        """
+        self._fetch_base_item = fetch_base_item
+        self._request = request
+
+    @abc.abstractmethod
+    async def get(self, collection_id: str) -> Dict[str, Any]:
+        """Return the base item for the collection and cache by collection id."""
+        pass
+
+
+class DefaultBaseItemCache(BaseItemCache):
+    """Implementation of the BaseItemCache that holds base items in a dict."""
+
+    def __init__(
+        self,
+        fetch_base_item: Callable[[str], Coroutine[Any, Any, Dict[str, Any]]],
+        request: Request,
+    ):
+        """Initialize the base item cache."""
+        self._base_items = {}
+        super().__init__(fetch_base_item, request)
+
+    async def get(self, collection_id: str):
+        """Return the base item for the collection and cache by collection id."""
+        if collection_id not in self._base_items:
+            self._base_items[collection_id] = await self._fetch_base_item(
+                collection_id,
+            )
+        return self._base_items[collection_id]