Add changes to opensearch.

Author: rhysrevans3

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -273,9 +273,7 @@ async def create_item_index(collection_id: str):
     }
 
     try:
-        await client.indices.create(
-            index=f"{index_by_collection_id(collection_id)}-000001", body=search_body
-        )
+        await client.indices.create(index=f"{index_by_collection_id(collection_id)}-000001", body=search_body)
     except TransportError as e:
         if e.status_code == 400:
             pass  # Ignore 400 status codes
@@ -356,12 +354,8 @@ class DatabaseLogic:
     client = AsyncSearchSettings().create_client
     sync_client = SyncSearchSettings().create_client
 
-    item_serializer: Type[serializers.ItemSerializer] = attr.ib(
-        default=serializers.ItemSerializer
-    )
-    collection_serializer: Type[serializers.CollectionSerializer] = attr.ib(
-        default=serializers.CollectionSerializer
-    )
+    item_serializer: Type[serializers.ItemSerializer] = attr.ib(default=serializers.ItemSerializer)
+    collection_serializer: Type[serializers.CollectionSerializer] = attr.ib(default=serializers.CollectionSerializer)
 
     extensions: List[str] = attr.ib(default=attr.Factory(list))
 
@@ -395,15 +389,9 @@ class DatabaseLogic:
                 "size": 10000,
             }
         },
-        "sun_elevation_frequency": {
-            "histogram": {"field": "properties.view:sun_elevation", "interval": 5}
-        },
-        "sun_azimuth_frequency": {
-            "histogram": {"field": "properties.view:sun_azimuth", "interval": 5}
-        },
-        "off_nadir_frequency": {
-            "histogram": {"field": "properties.view:off_nadir", "interval": 5}
-        },
+        "sun_elevation_frequency": {"histogram": {"field": "properties.view:sun_elevation", "interval": 5}},
+        "sun_azimuth_frequency": {"histogram": {"field": "properties.view:sun_azimuth", "interval": 5}},
+        "off_nadir_frequency": {"histogram": {"field": "properties.view:off_nadir", "interval": 5}},
         "centroid_geohash_grid_frequency": {
             "geohash_grid": {
                 "field": "properties.proj:centroid",
@@ -506,9 +494,7 @@ async def get_one_item(self, collection_id: str, item_id: str) -> Dict:
                 id=mk_item_id(item_id, collection_id),
             )
         except exceptions.NotFoundError:
-            raise NotFoundError(
-                f"Item {item_id} does not exist in Collection {collection_id}"
-            )
+            raise NotFoundError(f"Item {item_id} does not exist in Collection {collection_id}")
         return item["_source"]
 
     @staticmethod
@@ -531,9 +517,7 @@ def apply_free_text_filter(search: Search, free_text_queries: Optional[List[str]
         """Database logic to perform query for search endpoint."""
         if free_text_queries is not None:
             free_text_query_string = '" OR properties.\\*:"'.join(free_text_queries)
-            search = search.query(
-                "query_string", query=f'properties.\\*:"{free_text_query_string}"'
-            )
+            search = search.query("query_string", query=f'properties.\\*:"{free_text_query_string}"')
 
         return search
 
@@ -549,16 +533,10 @@ def apply_datetime_filter(search: Search, datetime_search):
             Search: The filtered search object.
         """
         if "eq" in datetime_search:
-            search = search.filter(
-                "term", **{"properties__datetime": datetime_search["eq"]}
-            )
+            search = search.filter("term", **{"properties__datetime": datetime_search["eq"]})
         else:
-            search = search.filter(
-                "range", properties__datetime={"lte": datetime_search["lte"]}
-            )
-            search = search.filter(
-                "range", properties__datetime={"gte": datetime_search["gte"]}
-            )
+            search = search.filter("range", properties__datetime={"lte": datetime_search["lte"]})
+            search = search.filter("range", properties__datetime={"gte": datetime_search["gte"]})
         return search
 
     @staticmethod
@@ -761,11 +739,7 @@ async def execute_search(
             if hits and (sort_array := hits[limit - 1].get("sort")):
                 next_token = urlsafe_b64encode(json.dumps(sort_array).encode()).decode()
 
-        matched = (
-            es_response["hits"]["total"]["value"]
-            if es_response["hits"]["total"]["relation"] == "eq"
-            else None
-        )
+        matched = es_response["hits"]["total"]["value"] if es_response["hits"]["total"]["relation"] == "eq" else None
         if count_task.done():
             try:
                 matched = count_task.result().get("count")
@@ -843,9 +817,7 @@ async def check_collection_exists(self, collection_id: str):
         if not await self.client.exists(index=COLLECTIONS_INDEX, id=collection_id):
             raise NotFoundError(f"Collection {collection_id} does not exist")
 
-    async def prep_create_item(
-        self, item: Item, base_url: str, exist_ok: bool = False
-    ) -> Item:
+    async def prep_create_item(self, item: Item, base_url: str, exist_ok: bool = False) -> Item:
         """
         Preps an item for insertion into the database.
 
@@ -867,15 +839,11 @@ async def prep_create_item(
             index=index_alias_by_collection_id(item["collection"]),
             id=mk_item_id(item["id"], item["collection"]),
         ):
-            raise ConflictError(
-                f"Item {item['id']} in collection {item['collection']} already exists"
-            )
+            raise ConflictError(f"Item {item['id']} in collection {item['collection']} already exists")
 
         return self.item_serializer.stac_to_db(item, base_url)
 
-    def sync_prep_create_item(
-        self, item: Item, base_url: str, exist_ok: bool = False
-    ) -> Item:
+    def sync_prep_create_item(self, item: Item, base_url: str, exist_ok: bool = False) -> Item:
         """
         Prepare an item for insertion into the database.
 
@@ -904,9 +872,7 @@ def sync_prep_create_item(
             index=index_alias_by_collection_id(collection_id),
             id=mk_item_id(item_id, collection_id),
         ):
-            raise ConflictError(
-                f"Item {item_id} in collection {collection_id} already exists"
-            )
+            raise ConflictError(f"Item {item_id} in collection {collection_id} already exists")
 
         return self.item_serializer.stac_to_db(item, base_url)
 
@@ -934,9 +900,7 @@ async def create_item(self, item: Item, refresh: bool = False):
         )
 
         if (meta := es_resp.get("meta")) and meta.get("status") == 409:
-            raise ConflictError(
-                f"Item {item_id} in collection {collection_id} already exists"
-            )
+            raise ConflictError(f"Item {item_id} in collection {collection_id} already exists")
 
     async def merge_patch_item(
         self,
@@ -993,59 +957,63 @@ async def json_patch_item(
         script_operations = []
 
         for operation in operations:
-            if operation["op"] in ["add", "replace"]:
-                if (
-                    operation["path"] == "collection"
-                    and collection_id != operation["value"]
-                ):
-                    await self.check_collection_exists(collection_id=operation["value"])
-                    new_collection_id = operation["value"]
+            if operation.path in ["collection", "id"] and operation.op in [
+                "add",
+                "replace",
+            ]:
+
+                if operation.path == "collection" and collection_id != operation.value:
+                    await self.check_collection_exists(collection_id=operation.value)
+                    new_collection_id = operation.value
 
-                if operation["path"] == "id" and item_id != operation["value"]:
-                    new_item_id = operation["value"]
+                if operation.path == "id" and item_id != operation.value:
+                    new_item_id = operation.value
 
             else:
                 script_operations.append(operation)
 
         script = operations_to_script(script_operations)
 
-        if not new_collection_id and not new_item_id:
+        try:
             await self.client.update(
                 index=index_by_collection_id(collection_id),
                 id=mk_item_id(item_id, collection_id),
                 script=script,
-                refresh=refresh,
+                refresh=True,
             )
 
+        except exceptions.BadRequestError as exc:
+            raise KeyError(exc.info["error"]["caused_by"]["to_string"]) from exc
+
+        item = await self.get_one_item(collection_id, item_id)
+
         if new_collection_id:
             await self.client.reindex(
                 body={
-                    "dest": {"index": f"{ITEMS_INDEX_PREFIX}{operation['value']}"},
+                    "dest": {"index": f"{ITEMS_INDEX_PREFIX}{new_collection_id}"},
                     "source": {
                         "index": f"{ITEMS_INDEX_PREFIX}{collection_id}",
                         "query": {"term": {"id": {"value": item_id}}},
                     },
                     "script": {
                         "lang": "painless",
                         "source": (
-                            f"""ctx._id = ctx._id.replace('{collection_id}', '{operation["value"]}');"""
-                            f"""ctx._source.collection = '{operation["value"]}';"""
-                            + script
+                            f"""ctx._id = ctx._id.replace('{collection_id}', '{new_collection_id}');"""
+                            f"""ctx._source.collection = '{new_collection_id}';"""
                         ),
                     },
                 },
                 wait_for_completion=True,
-                refresh=False,
+                refresh=True,
             )
-
-        item = await self.get_one_item(collection_id, item_id)
+            item["collection"] = new_collection_id
 
         if new_item_id:
             item["id"] = new_item_id
             item = await self.prep_create_item(item=item, base_url=base_url)
             await self.create_item(item=item, refresh=False)
 
-        if new_item_id or new_collection_id:
+        if new_collection_id or new_item_id:
 
             await self.delete_item(
                 item_id=item_id,
@@ -1055,9 +1023,7 @@ async def json_patch_item(
 
         return item
 
-    async def delete_item(
-        self, item_id: str, collection_id: str, refresh: bool = False
-    ):
+    async def delete_item(self, item_id: str, collection_id: str, refresh: bool = False):
         """Delete a single item from the database.
 
         Args:
@@ -1075,9 +1041,7 @@ async def delete_item(
                 refresh=refresh,
             )
         except exceptions.NotFoundError:
-            raise NotFoundError(
-                f"Item {item_id} in collection {collection_id} not found"
-            )
+            raise NotFoundError(f"Item {item_id} in collection {collection_id} not found")
 
     async def create_collection(self, collection: Collection, refresh: bool = False):
         """Create a single collection in the database.
@@ -1124,17 +1088,13 @@ async def find_collection(self, collection_id: str) -> Collection:
             collection as a `Collection` object. If the collection is not found, a `NotFoundError` is raised.
         """
         try:
-            collection = await self.client.get(
-                index=COLLECTIONS_INDEX, id=collection_id
-            )
+            collection = await self.client.get(index=COLLECTIONS_INDEX, id=collection_id)
         except exceptions.NotFoundError:
             raise NotFoundError(f"Collection {collection_id} not found")
 
         return collection["_source"]
 
-    async def update_collection(
-        self, collection_id: str, collection: Collection, refresh: bool = False
-    ):
+    async def update_collection(self, collection_id: str, collection: Collection, refresh: bool = False):
         """Update a collection from the database.
 
         Args:
@@ -1254,9 +1214,7 @@ async def json_patch_collection(
         if new_collection_id:
             collection["id"] = new_collection_id
             collection["links"] = resolve_links([], base_url)
-            await self.update_collection(
-                collection_id=collection_id, collection=collection, refresh=False
-            )
+            await self.update_collection(collection_id=collection_id, collection=collection, refresh=False)
 
         return collection
 
@@ -1277,14 +1235,10 @@ async def delete_collection(self, collection_id: str, refresh: bool = False):
             function also calls `delete_item_index` to delete the index for the items in the collection.
         """
         await self.find_collection(collection_id=collection_id)
-        await self.client.delete(
-            index=COLLECTIONS_INDEX, id=collection_id, refresh=refresh
-        )
+        await self.client.delete(index=COLLECTIONS_INDEX, id=collection_id, refresh=refresh)
         await delete_item_index(collection_id)
 
-    async def bulk_async(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
-    ) -> None:
+    async def bulk_async(self, collection_id: str, processed_items: List[Item], refresh: bool = False) -> None:
         """Perform a bulk insert of items into the database asynchronously.
 
         Args:
@@ -1306,9 +1260,7 @@ async def bulk_async(
             raise_on_error=False,
         )
 
-    def bulk_sync(
-        self, collection_id: str, processed_items: List[Item], refresh: bool = False
-    ) -> None:
+    def bulk_sync(self, collection_id: str, processed_items: List[Item], refresh: bool = False) -> None:
         """Perform a bulk insert of items into the database synchronously.
 
         Args:

stac_fastapi/tests/api/test_api.py

@@ -28,7 +28,9 @@
     "POST /collections",
     "POST /collections/{collection_id}/items",
     "PUT /collections/{collection_id}",
+    "PATCH /collections/{collection_id}",
     "PUT /collections/{collection_id}/items/{item_id}",
+    "PATCH /collections/{collection_id}/items/{item_id}",
     "GET /aggregations",
     "GET /aggregate",
     "POST /aggregations",