fix: Add relations in OS STAC pagination

This commit adds missing relations to OS STAC pagination:
- `collections/{collection_id}/items`: added `collection` and `parent` relations to the endpoint.
- `search`, `collections/{collection_id}/items`, `collections`: added the `previous` relation when following the `next` link.

Author: Yuri Zmytrakov

stac_fastapi/core/stac_fastapi/core/core.py

@@ -238,7 +238,7 @@ async def all_collections(self, **kwargs) -> stac_types.Collections:
         limit = int(request.query_params.get("limit", os.getenv("STAC_ITEM_LIMIT", 10)))
         token = request.query_params.get("token")
 
-        collections, next_token = await self.database.get_all_collections(
+        collections, next_token, prev_token = await self.database.get_all_collections(
             token=token, limit=limit, request=request
         )
 
@@ -256,6 +256,10 @@ async def all_collections(self, **kwargs) -> stac_types.Collections:
             next_link = PagingLinks(next=next_token, request=request).link_next()
             links.append(next_link)
 
+        if prev_token:
+            prev_token = PagingLinks(prev=prev_token, request=request).link_previous()
+            links.append(prev_token)
+
         return stac_types.Collections(collections=collections, links=links)
 
     async def get_collection(
@@ -343,7 +347,7 @@ async def item_collection(
             search = self.database.apply_bbox_filter(search=search, bbox=bbox)
 
         limit = int(request.query_params.get("limit", os.getenv("STAC_ITEM_LIMIT", 10)))
-        items, maybe_count, next_token = await self.database.execute_search(
+        items, maybe_count, next_token, prev_token = await self.database.execute_search(
             search=search,
             limit=limit,
             sort=None,
@@ -356,7 +360,12 @@ async def item_collection(
             self.item_serializer.db_to_stac(item, base_url=base_url) for item in items
         ]
 
-        links = await PagingLinks(request=request, next=next_token).get_links()
+        links = await PagingLinks(
+            request=request,
+            next=next_token,
+            prev=prev_token,
+            collection_id=collection_id,
+        ).get_links()
 
         return stac_types.ItemCollection(
             type="FeatureCollection",
@@ -568,7 +577,7 @@ async def post_search(
         if search_request.limit:
             limit = search_request.limit
 
-        items, maybe_count, next_token = await self.database.execute_search(
+        items, maybe_count, next_token, prev_token = await self.database.execute_search(
             search=search,
             limit=limit,
             token=search_request.token,
@@ -593,7 +602,9 @@ async def post_search(
             )
             for item in items
         ]
-        links = await PagingLinks(request=request, next=next_token).get_links()
+        links = await PagingLinks(
+            request=request, next=next_token, prev=prev_token
+        ).get_links()
 
         return stac_types.ItemCollection(
             type="FeatureCollection",

stac_fastapi/core/stac_fastapi/core/models/links.py

@@ -179,6 +179,8 @@ class PagingLinks(BaseLinks):
     """Create links for paging."""
 
     next: Optional[str] = attr.ib(kw_only=True, default=None)
+    prev: Optional[str] = attr.ib(kw_only=True, default=None)
+    collection_id: Optional[str] = attr.ib(default=None)
 
     def link_next(self) -> Optional[Dict[str, Any]]:
         """Create link for next page."""
@@ -203,3 +205,48 @@ def link_next(self) -> Optional[Dict[str, Any]]:
                 }
 
         return None
+
+    def link_previous(self) -> Optional[Dict[str, Any]]:
+        """Create link for previous page."""
+        if not self.prev:
+            return None
+
+        method = self.request.method
+        if method == "GET":
+            href = merge_params(self.url, {"token": self.prev})
+            link = dict(
+                rel=Relations.previous.value,
+                type=MimeTypes.json.value,
+                method=method,
+                href=href,
+            )
+            return link
+        if method == "POST":
+            return {
+                "rel": Relations.previous,
+                "type": MimeTypes.json.value,
+                "method": method,
+                "href": f"{self.request.url}",
+                "body": {**self.request.postbody, "token": self.prev},
+            }
+        return None
+
+    def link_parent(self) -> Optional[Dict[str, Any]]:
+        """Create the `parent` link if collection_id is provided."""
+        if self.collection_id:
+            return dict(
+                rel=Relations.parent.value,
+                type=MimeTypes.json.value,
+                href=urljoin(self.base_url, f"collections/{self.collection_id}"),
+            )
+        return None
+
+    def link_collection(self) -> Optional[Dict[str, Any]]:
+        """Create the `collection` link if collection_id is provided."""
+        if self.collection_id:
+            return dict(
+                rel=Relations.collection.value,
+                type=MimeTypes.json.value,
+                href=urljoin(self.base_url, f"collections/{self.collection_id}"),
+            )
+        return None

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -171,7 +171,7 @@ def __attrs_post_init__(self):
 
     async def get_all_collections(
         self, token: Optional[str], limit: int, request: Request
-    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+    ) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[str]]:
         """Retrieve a list of all collections from Elasticsearch, supporting pagination.
 
         Args:
@@ -189,12 +189,17 @@ async def get_all_collections(
             index=COLLECTIONS_INDEX,
             body={
                 "sort": [{"id": {"order": "asc"}}],
-                "size": limit,
+                "size": limit + 1,
                 **({"search_after": search_after} if search_after is not None else {}),
             },
         )
 
         hits = response["hits"]["hits"]
+
+        has_more = len(hits) > limit
+        if has_more:
+            hits = hits[:limit]
+
         collections = [
             self.collection_serializer.db_to_stac(
                 collection=hit["_source"], request=request, extensions=self.extensions
@@ -206,7 +211,11 @@ async def get_all_collections(
         if len(hits) == limit:
             next_token = hits[-1]["sort"][0]
 
-        return collections, next_token
+        prev_token = None
+        if token:
+            prev_token = token
+
+        return collections, next_token, prev_token
 
     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         """Retrieve a single item from the database.
@@ -506,7 +515,7 @@ async def execute_search(
         collection_ids: Optional[List[str]],
         datetime_search: Dict[str, Optional[str]],
         ignore_unavailable: bool = True,
-    ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str]]:
+    ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str], Optional[str]]:
         """Execute a search query with limit and other optional parameters.
 
         Args:
@@ -579,6 +588,10 @@ async def execute_search(
             if hits and (sort_array := hits[limit - 1].get("sort")):
                 next_token = urlsafe_b64encode(orjson.dumps(sort_array)).decode()
 
+        prev_token = None
+        if token and hits:
+            if hits and (first_item_sort := hits[0].get("sort")):
+                prev_token = urlsafe_b64encode(orjson.dumps(first_item_sort)).decode()
         matched = (
             es_response["hits"]["total"]["value"]
             if es_response["hits"]["total"]["relation"] == "eq"
@@ -590,7 +603,7 @@ async def execute_search(
             except Exception as e:
                 logger.error(f"Count task failed: {e}")
 
-        return items, matched, next_token
+        return items, matched, next_token, prev_token
 
     """ AGGREGATE LOGIC """
 

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -154,7 +154,7 @@ def __attrs_post_init__(self):
 
     async def get_all_collections(
         self, token: Optional[str], limit: int, request: Request
-    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+    ) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[str]]:
         """
         Retrieve a list of all collections from Opensearch, supporting pagination.
 
@@ -167,12 +167,16 @@ async def get_all_collections(
         """
         search_body = {
             "sort": [{"id": {"order": "asc"}}],
-            "size": limit,
+            "size": limit + 1,
         }
 
-        # Only add search_after to the query if token is not None and not empty
+        search_after = None
         if token:
-            search_after = [token]
+            try:
+                search_after = orjson.loads(urlsafe_b64decode(token))
+            except (ValueError, TypeError, orjson.JSONDecodeError):
+                search_after = [token]
+        if search_after:
             search_body["search_after"] = search_after
 
         response = await self.client.search(
@@ -181,6 +185,9 @@ async def get_all_collections(
         )
 
         hits = response["hits"]["hits"]
+        has_more = len(hits) > limit
+        if has_more:
+            hits = hits[:limit]
         collections = [
             self.collection_serializer.db_to_stac(
                 collection=hit["_source"], request=request, extensions=self.extensions
@@ -189,13 +196,21 @@ async def get_all_collections(
         ]
 
         next_token = None
-        if len(hits) == limit:
+        if has_more and hits:
             # Ensure we have a valid sort value for next_token
             next_token_values = hits[-1].get("sort")
             if next_token_values:
-                next_token = next_token_values[0]
+                next_token = urlsafe_b64encode(orjson.dumps(next_token_values)).decode()
 
-        return collections, next_token
+        prev_token = None
+        if token and hits:
+            first_token_values = hits[0].get("sort")
+            if first_token_values:
+                prev_token = urlsafe_b64encode(
+                    orjson.dumps(first_token_values)
+                ).decode()
+
+        return collections, next_token, prev_token
 
     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         """Retrieve a single item from the database.
@@ -495,7 +510,7 @@ async def execute_search(
         collection_ids: Optional[List[str]],
         datetime_search: Dict[str, Optional[str]],
         ignore_unavailable: bool = True,
-    ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str]]:
+    ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str], Optional[str]]:
         """Execute a search query with limit and other optional parameters.
 
         Args:
@@ -514,6 +529,7 @@ async def execute_search(
                 - The total number of results (if the count could be computed), or None if the count could not be
                 computed.
                 - The token to be used to retrieve the next set of results, or None if there are no more results.
+                - The token to be used to retrieve the previous set of results, or None if this is the first page.
 
         Raises:
             NotFoundError: If the collections specified in `collection_ids` do not exist.
@@ -570,10 +586,15 @@ async def execute_search(
         items = (hit["_source"] for hit in hits[:limit])
 
         next_token = None
+        prev_token = None
         if len(hits) > limit and limit < max_result_window:
             if hits and (sort_array := hits[limit - 1].get("sort")):
                 next_token = urlsafe_b64encode(orjson.dumps(sort_array)).decode()
 
+        if token and hits:
+            if hits and (first_item_sort := hits[0].get("sort")):
+                prev_token = urlsafe_b64encode(orjson.dumps(first_item_sort)).decode()
+
         matched = (
             es_response["hits"]["total"]["value"]
             if es_response["hits"]["total"]["relation"] == "eq"
@@ -585,7 +606,7 @@ async def execute_search(
             except Exception as e:
                 logger.error(f"Count task failed: {e}")
 
-        return items, matched, next_token
+        return items, matched, next_token, prev_token
 
     """ AGGREGATE LOGIC """