Merge pull request #91 from stac-utils/db_to_stac

Move db to stac serializer to core.py

Author: jonhealy1

CHANGELOG.md

@@ -22,6 +22,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Elasticsearch index mappings updated to be more thorough.
 - Endpoints that return items (e.g., /search) now sort the results by 'properties.datetime,id,collection'.
   Previously, there was no sort order defined.
+- Db_to_stac serializer moved to core.py for consistency as it existed in both core and database_logic previously. 
+- Use genexp in execute_search and get_all_collections to return results.
+- Added db_to_stac serializer to item_collection method in core.py.
 
 ### Removed
 

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/core.py

@@ -52,7 +52,11 @@ class CoreClient(AsyncBaseCoreClient):
     async def all_collections(self, **kwargs) -> Collections:
         """Read all collections from the database."""
         base_url = str(kwargs["request"].base_url)
-        collection_list = await self.database.get_all_collections(base_url=base_url)
+        collection_list = await self.database.get_all_collections()
+        collection_list = [
+            self.collection_serializer.db_to_stac(c, base_url=base_url)
+            for c in collection_list
+        ]
 
         links = [
             {
@@ -87,7 +91,7 @@ async def item_collection(
     ) -> ItemCollection:
         """Read an item collection from the database."""
         request: Request = kwargs["request"]
-        base_url = str(request.base_url)
+        base_url = str(kwargs["request"].base_url)
 
         items, maybe_count, next_token = await self.database.execute_search(
             search=self.database.apply_collections_filter(
@@ -96,9 +100,12 @@ async def item_collection(
             limit=limit,
             token=token,
             sort=None,
-            base_url=base_url,
         )
 
+        items = [
+            self.item_serializer.db_to_stac(item, base_url=base_url) for item in items
+        ]
+
         context_obj = None
         if self.extension_is_enabled("ContextExtension"):
             context_obj = {
@@ -269,9 +276,12 @@ async def post_search(
             limit=limit,
             token=search_request.token,  # type: ignore
             sort=sort,
-            base_url=base_url,
         )
 
+        items = [
+            self.item_serializer.db_to_stac(item, base_url=base_url) for item in items
+        ]
+
         # if self.extension_is_enabled("FieldsExtension"):
         #     if search_request.query is not None:
         #         query_include: Set[str] = set(

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -2,7 +2,7 @@
 import asyncio
 import logging
 from base64 import urlsafe_b64decode, urlsafe_b64encode
-from typing import Dict, List, Optional, Tuple, Type, Union
+from typing import Any, Dict, Iterable, List, Optional, Tuple, Type, Union
 
 import attr
 import elasticsearch
@@ -66,16 +66,12 @@ class DatabaseLogic:
 
     """CORE LOGIC"""
 
-    async def get_all_collections(self, base_url: str) -> List[Collection]:
+    async def get_all_collections(self) -> List[Dict[str, Any]]:
         """Database logic to retrieve a list of all collections."""
         # https://github.com/stac-utils/stac-fastapi-elasticsearch/issues/65
         # collections should be paginated, but at least return more than the default 10 for now
         collections = await self.client.search(index=COLLECTIONS_INDEX, size=1000)
-
-        return [
-            self.collection_serializer.db_to_stac(c["_source"], base_url=base_url)
-            for c in collections["hits"]["hits"]
-        ]
+        return (c["_source"] for c in collections["hits"]["hits"])
 
     async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
         """Database logic to retrieve a single item."""
@@ -194,8 +190,7 @@ async def execute_search(
         limit: int,
         token: Optional[str],
         sort: Optional[Dict[str, Dict[str, str]]],
-        base_url: str,
-    ) -> Tuple[List[Item], Optional[int], Optional[str]]:
+    ) -> Tuple[Iterable[Dict[str, Any]], Optional[int], Optional[str]]:
         """Database logic to execute search with limit."""
         search_after = None
         if token:
@@ -220,10 +215,7 @@ async def execute_search(
         es_response = await search_task
 
         hits = es_response["hits"]["hits"]
-        items = [
-            self.item_serializer.db_to_stac(hit["_source"], base_url=base_url)
-            for hit in hits
-        ]
+        items = (hit["_source"] for hit in hits)
 
         next_token = None
         if hits and (sort_array := hits[-1].get("sort")):