Merge branch 'main' into sfes-mongo

merge main

Author: jonhealy1

CHANGELOG.md

@@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Added
 
+- Added explicit mapping for ID in `ES_COLLECTIONS_MAPPINGS` [#198](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/198)
+
 ### Changed
 
+- Removed database logic from core.py all_collections [#196](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/196)
+- Changed OpenSearch config ssl_version to SSLv23 [#200](https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/pull/200)
+
 ### Fixed
 
 ## [v2.0.0]

stac_fastapi/core/stac_fastapi/core/core.py

@@ -238,7 +238,9 @@ async def get_collection(self, collection_id: str, **kwargs) -> Collection:
         """
         base_url = str(kwargs["request"].base_url)
         collection = await self.database.find_collection(collection_id=collection_id)
-        return self.collection_serializer.db_to_stac(collection, base_url)
+        return self.collection_serializer.db_to_stac(
+            collection=collection, base_url=base_url
+        )
 
     async def item_collection(
         self,

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -132,6 +132,7 @@
     "numeric_detection": False,
     "dynamic_templates": ES_MAPPINGS_DYNAMIC_TEMPLATES,
     "properties": {
+        "id": {"type": "keyword"},
         "extent.spatial.bbox": {"type": "long"},
         "extent.temporal.interval": {"type": "date"},
         "providers": {"type": "object", "enabled": False},
@@ -290,33 +291,43 @@ class DatabaseLogic:
     """CORE LOGIC"""
 
     async def get_all_collections(
-        self, token: Optional[str], limit: int
-    ) -> Iterable[Dict[str, Any]]:
-        """Retrieve a list of all collections from the database.
+        self, token: Optional[str], limit: int, base_url: str
+    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+        """Retrieve a list of all collections from Elasticsearch, supporting pagination.
 
         Args:
-            token (Optional[str]): The token used to return the next set of results.
-            limit (int): Number of results to return
+            token (Optional[str]): The pagination token.
+            limit (int): The number of results to return.
 
         Returns:
-            collections (Iterable[Dict[str, Any]]): A list of dictionaries containing the source data for each collection.
-
-        Notes:
-            The collections are retrieved from the Elasticsearch database using the `client.search` method,
-            with the `COLLECTIONS_INDEX` as the target index and `size=limit` to retrieve records.
-            The result is a generator of dictionaries containing the source data for each collection.
+            A tuple of (collections, next pagination token if any).
         """
         search_after = None
         if token:
-            search_after = urlsafe_b64decode(token.encode()).decode().split(",")
-        collections = await self.client.search(
+            search_after = [token]
+
+        response = await self.client.search(
             index=COLLECTIONS_INDEX,
-            search_after=search_after,
-            size=limit,
-            sort={"id": {"order": "asc"}},
+            body={
+                "sort": [{"id": {"order": "asc"}}],
+                "size": limit,
+                "search_after": search_after,
+            },
         )
-        hits = collections["hits"]["hits"]
-        return hits
+
+        hits = response["hits"]["hits"]
+        collections = [
+            self.collection_serializer.db_to_stac(
+                collection=hit["_source"], base_url=base_url
+            )
+            for hit in hits
+        ]
+
+        next_token = None
+        if len(hits) == limit:
+            next_token = hits[-1]["sort"][0]
+
+        return collections, next_token
 
     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         """Retrieve a single item from the database.

stac_fastapi/opensearch/stac_fastapi/opensearch/config.py

@@ -27,7 +27,7 @@ def _es_config() -> Dict[str, Any]:
         return config
 
     # Include SSL settings if using https
-    config["ssl_version"] = ssl.TLSVersion.TLSv1_3  # type: ignore
+    config["ssl_version"] = ssl.PROTOCOL_SSLv23  # type: ignore
     config["verify_certs"] = os.getenv("ES_VERIFY_CERTS", "true").lower() != "false"  # type: ignore
 
     # Include CA Certificates if verifying certs

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -134,6 +134,7 @@
     "numeric_detection": False,
     "dynamic_templates": ES_MAPPINGS_DYNAMIC_TEMPLATES,
     "properties": {
+        "id": {"type": "keyword"},
         "extent.spatial.bbox": {"type": "long"},
         "extent.temporal.interval": {"type": "date"},
         "providers": {"type": "object", "enabled": False},
@@ -311,36 +312,49 @@ class DatabaseLogic:
     """CORE LOGIC"""
 
     async def get_all_collections(
-        self,
-        token: Optional[str],
-        limit: int,
-    ) -> Iterable[Dict[str, Any]]:
-        """Retrieve a list of all collections from the database.
+        self, token: Optional[str], limit: int, base_url: str
+    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+        """
+        Retrieve a list of all collections from Opensearch, supporting pagination.
 
         Args:
-            token (Optional[str]): The token used to return the next set of results.
-            limit (int): Number of results to return
+            token (Optional[str]): The pagination token.
+            limit (int): The number of results to return.
 
         Returns:
-            collections (Iterable[Dict[str, Any]]): A list of dictionaries containing the source data for each collection.
-
-        Notes:
-            The collections are retrieved from the Elasticsearch database using the `client.search` method,
-            with the `COLLECTIONS_INDEX` as the target index and `size=limit` to retrieve records.
-            The result is a generator of dictionaries containing the source data for each collection.
+            A tuple of (collections, next pagination token if any).
         """
-        search_body: Dict[str, Any] = {}
+        search_body = {
+            "sort": [{"id": {"order": "asc"}}],
+            "size": limit,
+        }
+
+        # Only add search_after to the query if token is not None and not empty
         if token:
-            search_after = urlsafe_b64decode(token.encode()).decode().split(",")
+            search_after = [token]
             search_body["search_after"] = search_after
 
-        search_body["sort"] = {"id": {"order": "asc"}}
-
-        collections = await self.client.search(
-            index=COLLECTIONS_INDEX, body=search_body, size=limit
+        response = await self.client.search(
+            index="collections",
+            body=search_body,
         )
-        hits = collections["hits"]["hits"]
-        return hits
+
+        hits = response["hits"]["hits"]
+        collections = [
+            self.collection_serializer.db_to_stac(
+                collection=hit["_source"], base_url=base_url
+            )
+            for hit in hits
+        ]
+
+        next_token = None
+        if len(hits) == limit:
+            # 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]
+
+        return collections, next_token
 
     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         """Retrieve a single item from the database.