Adding patch endpoints to transactions extension to elasticsearch.

Author: rhysrevans3

stac_fastapi/core/stac_fastapi/core/base_database_logic.py

@@ -1,7 +1,7 @@
 """Base database logic."""
 
 import abc
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 
 class BaseDatabaseLogic(abc.ABC):
@@ -29,6 +29,30 @@ async def create_item(self, item: Dict, refresh: bool = False) -> None:
         """Create an item in the database."""
         pass
 
+    @abc.abstractmethod
+    async def merge_patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        item: Dict,
+        base_url: str,
+        refresh: bool = True,
+    ) -> Dict:
+        """Patch a item in the database follows RF7396."""
+        pass
+
+    @abc.abstractmethod
+    async def json_patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        operations: List,
+        base_url: str,
+        refresh: bool = True,
+    ) -> Dict:
+        """Patch a item in the database follows RF6902."""
+        pass
+
     @abc.abstractmethod
     async def delete_item(
         self, item_id: str, collection_id: str, refresh: bool = False
@@ -41,6 +65,28 @@ async def create_collection(self, collection: Dict, refresh: bool = False) -> No
         """Create a collection in the database."""
         pass
 
+    @abc.abstractmethod
+    async def merge_patch_collection(
+        self,
+        collection_id: str,
+        collection: Dict,
+        base_url: str,
+        refresh: bool = True,
+    ) -> Dict:
+        """Patch a collection in the database follows RF7396."""
+        pass
+
+    @abc.abstractmethod
+    async def json_patch_collection(
+        self,
+        collection_id: str,
+        operations: List,
+        base_url: str,
+        refresh: bool = True,
+    ) -> Dict:
+        """Patch a collection in the database follows RF6902."""
+        pass
+
     @abc.abstractmethod
     async def find_collection(self, collection_id: str) -> Dict:
         """Find a collection in the database."""

stac_fastapi/core/stac_fastapi/core/core.py

@@ -1,4 +1,5 @@
 """Core client."""
+
 import logging
 import re
 from datetime import datetime as datetime_type
@@ -708,6 +709,58 @@ async def update_item(
 
         return ItemSerializer.db_to_stac(item, base_url)
 
+    @overrides
+    async def merge_patch_item(
+        self, collection_id: str, item_id: str, item: stac_types.PartialItem, **kwargs
+    ) -> Optional[stac_types.Item]:
+        """Patch an item in the collection following RF7396..
+
+        Args:
+            collection_id (str): The ID of the collection the item belongs to.
+            item_id (str): The ID of the item to be updated.
+            item (stac_types.PartialItem): The partial item data.
+            kwargs: Other optional arguments, including the request object.
+
+        Returns:
+            stac_types.Item: The patched item object.
+
+        """
+        item = await self.database.merge_patch_item(
+            collection_id=collection_id,
+            item_id=item_id,
+            item=item,
+            base_url=str(kwargs["request"].base_url),
+        )
+        return ItemSerializer.db_to_stac(item, base_url=str(kwargs["request"].base_url))
+
+    @overrides
+    async def json_patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        operations: List[stac_types.PatchOperation],
+        **kwargs,
+    ) -> Optional[stac_types.Item]:
+        """Patch an item in the collection following RF6902.
+
+        Args:
+            collection_id (str): The ID of the collection the item belongs to.
+            item_id (str): The ID of the item to be updated.
+            operations (List): List of operations to run on item.
+            kwargs: Other optional arguments, including the request object.
+
+        Returns:
+            stac_types.Item: The patched item object.
+
+        """
+        item = await self.database.json_patch_item(
+            collection_id=collection_id,
+            item_id=item_id,
+            base_url=str(kwargs["request"].base_url),
+            operations=operations,
+        )
+        return ItemSerializer.db_to_stac(item, base_url=str(kwargs["request"].base_url))
+
     @overrides
     async def delete_item(
         self, item_id: str, collection_id: str, **kwargs
@@ -788,6 +841,59 @@ async def update_collection(
             extensions=[type(ext).__name__ for ext in self.database.extensions],
         )
 
+    @overrides
+    async def merge_patch_collection(
+        self, collection_id: str, collection: stac_types.PartialCollection, **kwargs
+    ) -> Optional[stac_types.Collection]:
+        """Patch a collection following RF7396..
+
+        Args:
+            collection_id (str): The ID of the collection  to patch.
+            collection (stac_types.Collection): The partial collection data.
+            kwargs: Other optional arguments, including the request object.
+
+        Returns:
+            stac_types.Collection: The patched collection object.
+
+        """
+        collection = await self.database.merge_patch_collection(
+            collection_id=collection_id,
+            base_url=str(kwargs["request"].base_url),
+            collection=collection,
+        )
+
+        return CollectionSerializer.db_to_stac(
+            collection,
+            kwargs["request"],
+            extensions=[type(ext).__name__ for ext in self.database.extensions],
+        )
+
+    @overrides
+    async def json_patch_collection(
+        self, collection_id: str, operations: List[stac_types.PatchOperation], **kwargs
+    ) -> Optional[stac_types.Collection]:
+        """Patch a collection following RF6902.
+
+        Args:
+            collection_id (str): The ID of the collection to patch.
+            operations (List): List of operations to run on collection.
+            kwargs: Other optional arguments, including the request object.
+
+        Returns:
+            stac_types.Collection: The patched collection object.
+
+        """
+        collection = await self.database.json_patch_collection(
+            collection_id=collection_id,
+            operations=operations,
+            base_url=str(kwargs["request"].base_url),
+        )
+        return CollectionSerializer.db_to_stac(
+            collection,
+            kwargs["request"],
+            extensions=[type(ext).__name__ for ext in self.database.extensions],
+        )
+
     @overrides
     async def delete_collection(
         self, collection_id: str, **kwargs

stac_fastapi/core/stac_fastapi/core/utilities.py

@@ -3,6 +3,8 @@
 This module contains functions for transforming geospatial coordinates,
 such as converting bounding boxes to polygon representations.
 """
+
+import json
 from typing import Any, Dict, List, Optional, Set, Union
 
 from stac_fastapi.types.stac import Item
@@ -133,3 +135,64 @@ def dict_deep_update(merge_to: Dict[str, Any], merge_from: Dict[str, Any]) -> No
             dict_deep_update(merge_to[k], merge_from[k])
         else:
             merge_to[k] = v
+
+
+def merge_to_operations(data: Dict) -> List:
+    """Convert merge operation to list of RF6902 operations.
+
+    Args:
+        data: dictionary to convert.
+
+    Returns:
+        List: list of RF6902 operations.
+    """
+    operations = []
+
+    for key, value in data.copy().items():
+
+        if value is None:
+            operations.append({"op": "remove", "path": key})
+            continue
+
+        elif isinstance(value, dict):
+            nested_operations = merge_to_operations(value)
+
+            for nested_operation in nested_operations:
+                nested_operation["path"] = f"{key}.{nested_operation['path']}"
+                operations.append(nested_operation)
+
+        else:
+            operations.append({"op": "add", "path": key, "value": value})
+
+    return operations
+
+
+def operations_to_script(operations: List) -> Dict:
+    """Convert list of operation to painless script.
+
+    Args:
+        operations: List of RF6902 operations.
+
+    Returns:
+        Dict: elasticsearch update script.
+    """
+    source = ""
+    for operation in operations:
+        if operation["op"] in ["copy", "move"]:
+            source += (
+                f"ctx._source.{operation['path']} = ctx._source.{operation['from']};"
+            )
+
+        if operation["op"] in ["remove", "move"]:
+            nest, partition, key = operation["path"].rpartition(".")
+            source += f"ctx._source.{nest + partition}remove('{key}');"
+
+        if operation["op"] in ["add", "replace"]:
+            source += (
+                f"ctx._source.{operation['path']} = {json.dumps(operation['value'])};"
+            )
+
+    return {
+        "source": source,
+        "lang": "painless",
+    }