Repository: stac-utils/stac-fastapi
Pull request #744 (2024-08-21 07:44:51)

Patch endpoints
**Description:**
Adds `PATCH` endpoints to transaction extension. Adds support for [RFC 6902](https://tools.ietf.org/html/rfc6902) and [RFC 7396](https://tools.ietf.org/html/rfc7396). Pivots on header `Content-Type` value.

Related pull request: https://github.com/stac-api-extensions/transaction/pull/14

**PR Checklist:**

- [x] `pre-commit` hooks pass locally
- [x] Tests pass (run `make test`)
- [x] Documentation has been updated to reflect changes, if applicable, and docs build successfully (run `make docs`)
- [x] Changes are added to the [CHANGELOG](https://github.com/stac-utils/stac-fastapi/blob/main/CHANGES.md).


rhysrevans3: I've had to switch from stac-pydantic `Item`/`Collection` to `Dict` to allow for partial updates. Not sure if this is the best method or if you can switch off pydantic validation. Another option would be having separate models for partial items/collections in stac-pydantic where all attributes.
vincentsarago: > I've had to switch from stac-pydantic Item/Collection to Dict to allow for partial updates. Not sure if this is the best method or if you can switch off pydantic validation. Another option would be having separate models for partial items/collections in stac-pydantic where all attributes.

Would a typedDict make more sense? 
vincentsarago: @rhysrevans3 should we wait for https://github.com/stac-api-extensions/transaction/pull/14 to be merged/published? 
vincentsarago: @rhysrevans3 I'm trying to get a general sense of the `patch` operation. It's not clear to me what are the requirements that an API should implement. By default it seems that a client will have to support a lot of things 

For example, I'm looking at https://docs.ogc.org/DRAFTS/20-002.html#update_clause and I don't see reference to the `move/copy` operations 
rhysrevans3: My understanding is [RFC 5789](https://datatracker.ietf.org/doc/html/rfc5789), which the [OGC doc](https://docs.ogc.org/DRAFTS/20-002.html#update_clause) references, say a PATCH endpoint should except a set of actions to modify the requested resource. But it doesn't specify what these actions should be.

[RFC 6902](https://datatracker.ietf.org/doc/html/rfc6902) extends this and specify a set of allowed operations (add, remove, replace, move, copy, test) which is what I have suggested in https://github.com/stac-api-extensions/transaction/pull/14.

Currently the [README](https://github.com/stac-api-extensions/transaction/blob/main/README.md?plain=1#L89) on the transaction extension suggests the PATCH endpoint follows [RFC 7396](https://datatracker.ietf.org/doc/html/rfc7396) which accepts a partial json document which it merges with the existing document this might not be the best choice for STAC because of it's handling of `null` values. As mentioned in this issue https://github.com/stac-api-extensions/transaction/issues/13
vincentsarago: @rhysrevans3 The overall looks great, just added one comment about using stac-pydantic base model.

As I've asked couple weeks ago, should we wait for https://github.com/stac-api-extensions/transaction/pull/14 to be merged? do you know anyone who could help move this PR forward? 
rhysrevans3: > @rhysrevans3 The overall looks great, just added one comment about using stac-pydantic base model.
> 
> As I've asked couple weeks ago, should we wait for [stac-api-extensions/transaction#14](https://github.com/stac-api-extensions/transaction/pull/14) to be merged? do you know anyone who could help move this PR forward?

Ideally yes I think it would be better to have that PR merged first.

But my understanding is that [RFC 6902](https://datatracker.ietf.org/doc/html/rfc6902) and [RFC 7396](https://datatracker.ietf.org/doc/html/rfc7396) are implementations of [RFC 5789](https://datatracker.ietf.org/doc/html/rfc5789), which is the current STAC/OGC specification. So this merging this pull still keeps within the current STAC specification.

But given @m-mohr comments on that pull request maybe it's best to wait for OGC's opinion.
m-mohr: Did someone open an issue with OGC API - Features? (If yes, where?) Otherwise you may wait forever ;) 
rhysrevans3: Sorry I've been at a conference/on leave I've created a new issue [here](https://github.com/opengeospatial/ogcapi-features/issues/957)
m-mohr: @rhysrevans3 Thanks.

Given the response in https://github.com/opengeospatial/ogcapi-features/issues/957, we may also want to open an issue in the Transaction extension itself so that we can reflect the updated to OGC API - Features in the extension, too. Would you mind opening that issue, too?
m-mohr: Oh cool, too many repos to monitor. Left my comment(s) there.
rhysrevans3: @vincentsarago [The transaction extension pull request](https://github.com/stac-api-extensions/transaction/issues/13) is just waiting for one more review. But I wondered if we could get this one merged in first? I'm confident the changes will be accept and this doesn't break the existing transaction extension.
vincentsarago: hi @rhysrevans3 

I'm a bit concerned that if merging this PR as is it would introduce a breaking change making `merge_patch_item`, `json_patch_item`, `merge_patch_collection` and `json_patch_collection` mandatory  methods in any Transaction client 

Also reading the spec 

> The default standard for PATCH is [RFC 7396](https://tools.ietf.org/html/rfc7396).
 Optionally, in addition to RFC 7396, [RFC 6902](https://tools.ietf.org/html/rfc6902) can be supported with the two methods
 being distinguished via the Accept/Content-Type header.

This sounds to me that only RFC7396 (merge) should be implemented and RFC6902 (patch) should be optional, but because of the 4 mandatory methods this is not reflected. 


I understand correctly, the current implementation is not following the transaction spec because we don't have `PATCH` support https://github.com/stac-utils/stac-fastapi/blob/20ae9cfaf87ed892ef3235d979892e7e24c63fc0/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py#L151-L154

Note: IMO we should move everything related to `transaction` to the `extension` module (https://github.com/stac-utils/stac-fastapi/blob/88639230e3bd1573e23e3a646204cd85932079cd/stac_fastapi/types/stac_fastapi/types/core.py#L40C1-L621C1). But it might be easier to do it after merging this PR 
rhysrevans3: @vincentsarago you're right I had missed that. I've updated `json_patch_item` and  `json_patch_collection` to raise an `NotImplementedError` as a default. Is that enough or does it need to be in the registration of the endpoint?

Happy to use this pull request to move everything to the `transaction` to the `extension module` if required.
rhysrevans3: I could also add a `NotImplemented` default to the `merge-patch` to not break existing implementations? I'm unsure if the Transaction extension _requires_ a PATCH endpoint or if you could specify a different `Content-Type`. For example if you wanted to implement `json-patch` and not `merge-patch` or even a non-JSON PATCH like [XML PATCH RFC 5261](https://datatracker.ietf.org/doc/html/rfc5261)
rhysrevans3: I've written a conversion from `merge` to `patch` for the elasticsearch-opensearch backend if this is useful to other backends? I could move it to the core utils?

https://github.com/stac-utils/stac-fastapi-elasticsearch-opensearch/blob/2b1381821ec978ea254df0c8fae87420186a8789/stac_fastapi/core/stac_fastapi/core/utilities.py#L145-L171
vincentsarago: @rhysrevans3 yeah that would be great. It will lower the barrier for people to support both patch and merge 🙏 

Maybe it could be a method on the PartialItem/PartialCollection model 

☝️ while writing this I realized that they are typedDict not pydantic model 🤔 
rhysrevans3: > @rhysrevans3 yeah that would be great. It will lower the barrier for people to support both patch and merge 🙏
> 
> Maybe it could be a method on the PartialItem/PartialCollection model
> 
> ☝️ while writing this I realized that they are typedDict not pydantic model 🤔

Yes I copied the Item/Collection models in the same file. I think the reason for using TypeDict over Pydantic models was performance (avoid validation) I remember there was some discussion when we moved to pydantic v2 #625 which suggested that's no longer an issue? Should I switch to using Pydantic models?
vincentsarago: >  Should I switch to using Pydantic models?

Yes we should use pydantic model for input validation. It's only for the `output` that we have optional validation 
rhysrevans3: > @rhysrevans3 I'm sorry it's taking longer 😅
> 
> I now think (with the addition of a `merge to patch` method) that by default we should make both patch/merge supported by default 😅.
> 
> On think I'm still not sure is how the merging/patching is made.
> 
> I'm looking at https://github.com/rhysrevans3/stac-fastapi-elasticsearch/blob/2b1381821ec978ea254df0c8fae87420186a8789/stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py#L940-L1031 to try to understand what's going on.
> 
> We should write some pseudo code in the doc (in the docstring first) to show how the `patch` method will look like
> 
> ```python
>     def patch_item(
>         self,
>         collection_id: str,
>         item_id: str,
>         patch: Union[stac.PartialItem, List[stac.PatchOperation]],
>         **kwargs,
>     ) -> Optional[Union[stac.Item, Response]]:
>         """Update an item from a collection.
> 
>         Called with `PATCH /collections/{collection_id}/items/{item_id}`
> 
>         Args:
>             item_id: id of the item.
>             collection_id: id of the collection.
>             patch: either the partial item or list of patch operations.
> 
>         Returns:
>             The patched item.
>         """
>         # convert patch item to list of merge operations
>         if isinstance(patch, PartialItem):
>             patch = merge_to_operations(patch)
> 
>         # Get Original Item
>         item = backend.get_item(collection_id, item_id)
> 
>         # Update Item body from merge op
>         item = update_item(item, patch)
> 
>         # Push Item to the backend
>         _ = backend.create_item(collection_id, item_id, item)
> 
>         return item
> ```

No worries I'd rather it takes a bit longer and we get it right!

For Elasticsearch you can update records using Java style scripts so for each operation type I've written an equivalent Elasticsearch script. I'm not sure if you could do the same for PostgreSQL or the other backends.

A merge is equivalent to a group of add and remove operations (null values are used to remove). so you can convert from merge to operations to Elasticsearch.
rhysrevans3: I've not tested this but this is an example of running a list patch operations on an item:

```python
from stac_fastapi.types.stac import PatchOperation
from stac_pydantic import Item

def patch_item(item: Item, operations: PatchOperation) -> Item:
    item_dict = item.model_dump()

    for operation in operations:
        path_parts = operation.path.split('/')
 
       if operation.op == "test":
            test_value = item_dict.copy()
            for path_part in path_parts:
                test_value = test_value[path_part]

            assert test_value == operation.value
            continue

        if operation.op == "replace":
            nest = item_dict.copy()
            for path_part in path_parts:
                assert path_part in nest
                nest = nest[path_part]

        update = {}

        if operation.op in ["add", "copy", "replace", "move"]:
            if operation.hasattr("from_"):
                from_parts = operation.from_.split('/')

                value = item_dict.copy()
                for path_part in from_parts:
                    value = value[path_part]

            else:
                value = item.value

            update = value
            for path_part in path_parts.reverse():
                update = {path_part: update}

        if operation.op in ["remove", "move"]:
            if operation.op == "move":
                path_parts = from_parts
 
            last_part = path_parts.pop(-1)

            nest = item_dict
            for path_part in path_parts:
                nest = nest[path_part]

            del nest[last_part]

        return Item.model_validate(item_dict | update)
```
vincentsarago: FYI: sorry for letting this PR stale. Because it adds breaking changes I'll wait for the next breaking change release to be planned before merging. 

rhysrevans3: @vincentsarago thanks for staying on top of this. Including it in the next major release sounds sensible 😃 

Patch:
diff --git a/CHANGES.md b/CHANGES.md
index 9c21410..98aa957 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -4,6 +4,7 @@
 
 ### Changed
 
+* Add Item and Collection `PATCH` endpoints with support for [RFC 6902](https://tools.ietf.org/html/rfc6902) and [RFC 7396](https://tools.ietf.org/html/rfc7386) in the `TransactionExtension`
 - remove support of `cql-json` in Filter extension ([#840](https://github.com/stac-utils/stac-fastapi/pull/840))
 
 ### Fixed
@@ -175,7 +176,6 @@
 ## [3.0.0] - 2024-07-29
 
 Full changelog: https://stac-utils.github.io/stac-fastapi/migrations/v3.0.0/#changelog
-
 **Changes since 3.0.0b3:**
 
 ### Changed
diff --git a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py
index 2287bc8..c0e390a 100644
--- a/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py
+++ b/stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py
@@ -5,6 +5,7 @@ from typing import List, Optional, Type, Union
 
 import attr
 from fastapi import APIRouter, Body, FastAPI
+from pydantic import TypeAdapter
 from stac_pydantic import Collection, Item, ItemCollection
 from stac_pydantic.shared import MimeTypes
 from starlette.responses import Response
@@ -15,6 +16,7 @@ from stac_fastapi.api.routes import create_async_endpoint
 from stac_fastapi.types.config import ApiSettings
 from stac_fastapi.types.core import AsyncBaseTransactionsClient, BaseTransactionsClient
 from stac_fastapi.types.extension import ApiExtension
+from stac_fastapi.types.stac import PartialCollection, PartialItem, PatchOperation
 
 
 class TransactionConformanceClasses(str, Enum):
@@ -42,6 +44,16 @@ class PutItem(ItemUri):
     item: Annotated[Item, Body()] = attr.ib(default=None)
 
 
+@attr.s
+class PatchItem(ItemUri):
+    """Patch Item."""
+
+    patch: Annotated[
+        Union[PartialItem, List[PatchOperation]],
+        Body(),
+    ] = attr.ib(default=None)
+
+
 @attr.s
 class PutCollection(CollectionUri):
     """Update Collection."""
@@ -49,6 +61,95 @@ class PutCollection(CollectionUri):
     collection: Annotated[Collection, Body()] = attr.ib(default=None)
 
 
+@attr.s
+class PatchCollection(CollectionUri):
+    """Patch Collection."""
+
+    patch: Annotated[
+        Union[PartialCollection, List[PatchOperation]],
+        Body(),
+    ] = attr.ib(default=None)
+
+
+_patch_item_schema = TypeAdapter(List[PatchOperation]).json_schema() | {
+    "examples": [
+        [
+            {
+                "op": "add",
+                "path": "/properties/foo",
+                "value": "bar",
+            },
+            {
+                "op": "replace",
+                "path": "/properties/foo",
+                "value": "bar",
+            },
+            {
+                "op": "test",
+                "path": "/properties/foo",
+                "value": "bar",
+            },
+            {
+                "op": "copy",
+                "path": "/properties/foo",
+                "from": "/properties/bar",
+            },
+            {
+                "op": "move",
+                "path": "/properties/foo",
+                "from": "/properties/bar",
+            },
+            {
+                "op": "remove",
+                "path": "/properties/foo",
+            },
+        ]
+    ]
+}
+# ref: https://github.com/pydantic/pydantic/issues/889
+_patch_item_schema["items"]["anyOf"] = list(_patch_item_schema["$defs"].values())
+
+_patch_collection_schema = TypeAdapter(List[PatchOperation]).json_schema() | {
+    "examples": [
+        [
+            {
+                "op": "add",
+                "path": "/summeries/foo",
+                "value": "bar",
+            },
+            {
+                "op": "replace",
+                "path": "/summeries/foo",
+                "value": "bar",
+            },
+            {
+                "op": "test",
+                "path": "/summeries/foo",
+                "value": "bar",
+            },
+            {
+                "op": "copy",
+                "path": "/summeries/foo",
+                "from": "/summeries/bar",
+            },
+            {
+                "op": "move",
+                "path": "/summeries/foo",
+                "from": "/summeries/bar",
+            },
+            {
+                "op": "remove",
+                "path": "/summeries/foo",
+            },
+        ]
+    ]
+}
+# ref: https://github.com/pydantic/pydantic/issues/889
+_patch_collection_schema["items"]["anyOf"] = list(
+    _patch_collection_schema["$defs"].values()
+)
+
+
 @attr.s
 class TransactionExtension(ApiExtension):
     """Transaction Extension.
@@ -126,6 +227,47 @@ class TransactionExtension(ApiExtension):
             endpoint=create_async_endpoint(self.client.update_item, PutItem),
         )
 
+    def register_patch_item(self):
+        """Register patch item endpoint (PATCH
+        /collections/{collection_id}/items/{item_id})."""
+        self.router.add_api_route(
+            name="Patch Item",
+            path="/collections/{collection_id}/items/{item_id}",
+            response_model=Item if self.settings.enable_response_models else None,
+            responses={
+                200: {
+                    "content": {
+                        MimeTypes.geojson.value: {},
+                    },
+                    "model": Item,
+                }
+            },
+            openapi_extra={
+                "requestBody": {
+                    "content": {
+                        "application/json-patch+json": {
+                            "schema": _patch_item_schema,
+                        },
+                        "application/merge-patch+json": {
+                            "schema": PartialItem.model_json_schema(),
+                        },
+                        "application/json": {
+                            "schema": PartialItem.model_json_schema(),
+                        },
+                    },
+                    "required": True,
+                },
+            },
+            response_class=self.response_class,
+            response_model_exclude_unset=True,
+            response_model_exclude_none=True,
+            methods=["PATCH"],
+            endpoint=create_async_endpoint(
+                self.client.patch_item,
+                PatchItem,
+            ),
+        )
+
     def register_delete_item(self):
         """Register delete item endpoint (DELETE
         /collections/{collection_id}/items/{item_id})."""
@@ -148,11 +290,6 @@ class TransactionExtension(ApiExtension):
             endpoint=create_async_endpoint(self.client.delete_item, ItemUri),
         )
 
-    def register_patch_item(self):
-        """Register patch item endpoint (PATCH
-        /collections/{collection_id}/items/{item_id})."""
-        raise NotImplementedError
-
     def register_create_collection(self):
         """Register create collection endpoint (POST /collections)."""
         self.router.add_api_route(
@@ -196,6 +333,46 @@ class TransactionExtension(ApiExtension):
             endpoint=create_async_endpoint(self.client.update_collection, PutCollection),
         )
 
+    def register_patch_collection(self):
+        """Register patch collection endpoint (PATCH /collections/{collection_id})."""
+        self.router.add_api_route(
+            name="Patch Collection",
+            path="/collections/{collection_id}",
+            response_model=Collection if self.settings.enable_response_models else None,
+            responses={
+                200: {
+                    "content": {
+                        MimeTypes.geojson.value: {},
+                    },
+                    "model": Collection,
+                }
+            },
+            openapi_extra={
+                "requestBody": {
+                    "content": {
+                        "application/json-patch+json": {
+                            "schema": _patch_collection_schema,
+                        },
+                        "application/merge-patch+json": {
+                            "schema": PartialCollection.model_json_schema(),
+                        },
+                        "application/json": {
+                            "schema": PartialCollection.model_json_schema(),
+                        },
+                    },
+                    "required": True,
+                },
+            },
+            response_class=self.response_class,
+            response_model_exclude_unset=True,
+            response_model_exclude_none=True,
+            methods=["PATCH"],
+            endpoint=create_async_endpoint(
+                self.client.patch_collection,
+                PatchCollection,
+            ),
+        )
+
     def register_delete_collection(self):
         """Register delete collection endpoint (DELETE /collections/{collection_id})."""
         self.router.add_api_route(
@@ -217,10 +394,6 @@ class TransactionExtension(ApiExtension):
             endpoint=create_async_endpoint(self.client.delete_collection, CollectionUri),
         )
 
-    def register_patch_collection(self):
-        """Register patch collection endpoint (PATCH /collections/{collection_id})."""
-        raise NotImplementedError
-
     def register(self, app: FastAPI) -> None:
         """Register the extension with a FastAPI application.
 
@@ -233,8 +406,10 @@ class TransactionExtension(ApiExtension):
         self.router.prefix = app.state.router_prefix
         self.register_create_item()
         self.register_update_item()
+        self.register_patch_item()
         self.register_delete_item()
         self.register_create_collection()
         self.register_update_collection()
+        self.register_patch_collection()
         self.register_delete_collection()
         app.include_router(self.router, tags=["Transaction Extension"])
diff --git a/stac_fastapi/types/stac_fastapi/types/core.py b/stac_fastapi/types/stac_fastapi/types/core.py
index c3d84e9..7d5e1fd 100644
--- a/stac_fastapi/types/stac_fastapi/types/core.py
+++ b/stac_fastapi/types/stac_fastapi/types/core.py
@@ -80,6 +80,37 @@ class BaseTransactionsClient(abc.ABC):
         """
         ...
 
+    @abc.abstractmethod
+    def patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        patch: Union[stac.PartialItem, List[stac.PatchOperation]],
+        **kwargs,
+    ) -> Optional[Union[stac.Item, Response]]:
+        """Update an item from a collection.
+
+        Called with `PATCH /collections/{collection_id}/items/{item_id}`
+
+        example:
+            # convert merge patch item to list of operations
+            if isinstance(patch, PartialItem):
+                patch = patch.operations()
+
+            item = backend.patch_item(collection_id, item_id, patch)
+
+            return item
+
+        Args:
+            item_id: id of the item.
+            collection_id: id of the collection.
+            patch: either the partial item or list of patch operations.
+
+        Returns:
+            The patched item.
+        """
+        ...
+
     @abc.abstractmethod
     def delete_item(
         self, item_id: str, collection_id: str, **kwargs
@@ -133,6 +164,35 @@ class BaseTransactionsClient(abc.ABC):
         """
         ...
 
+    @abc.abstractmethod
+    def patch_collection(
+        self,
+        collection_id: str,
+        patch: Union[stac.PartialCollection, List[stac.PatchOperation]],
+        **kwargs,
+    ) -> Optional[Union[stac.Collection, Response]]:
+        """Update a collection.
+
+        Called with `PATCH /collections/{collection_id}`
+
+        example:
+            # convert merge patch item to list of operations
+            if isinstance(patch, PartialCollection):
+                patch = patch.operations()
+
+            collection = backend.patch_collection(collection_id, patch)
+
+            return collection
+
+        Args:
+            collection_id: id of the collection.
+            patch: either the partial collection or list of patch operations.
+
+        Returns:
+            The patched collection.
+        """
+        ...
+
     @abc.abstractmethod
     def delete_collection(
         self, collection_id: str, **kwargs
@@ -193,6 +253,37 @@ class AsyncBaseTransactionsClient(abc.ABC):
         """
         ...
 
+    @abc.abstractmethod
+    async def patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        patch: Union[stac.PartialItem, List[stac.PatchOperation]],
+        **kwargs,
+    ) -> Optional[Union[stac.Item, Response]]:
+        """Update an item from a collection.
+
+        Called with `PATCH /collections/{collection_id}/items/{item_id}`
+
+        example:
+            # convert merge patch item to list of operations
+            if isinstance(patch, PartialItem):
+                patch = patch.operations()
+
+            item = backend.patch_item(collection_id, item_id, patch)
+
+            return item
+
+        Args:
+            item_id: id of the item.
+            collection_id: id of the collection.
+            patch: either the partial item or list of patch operations.
+
+        Returns:
+            The patched item.
+        """
+        ...
+
     @abc.abstractmethod
     async def delete_item(
         self, item_id: str, collection_id: str, **kwargs
@@ -246,6 +337,35 @@ class AsyncBaseTransactionsClient(abc.ABC):
         """
         ...
 
+    @abc.abstractmethod
+    async def patch_collection(
+        self,
+        collection_id: str,
+        patch: Union[stac.PartialCollection, List[stac.PatchOperation]],
+        **kwargs,
+    ) -> Optional[Union[stac.Collection, Response]]:
+        """Update a collection.
+
+        Called with `PATCH /collections/{collection_id}`
+
+        example:
+            # convert merge patch item to list of operations
+            if isinstance(patch, PartialCollection):
+                patch = patch.operations()
+
+            collection = backend.patch_collection(collection_id, patch)
+
+            return collection
+
+        Args:
+            collection_id: id of the collection.
+            patch: either the partial collection or list of patch operations.
+
+        Returns:
+            The patched collection.
+        """
+        ...
+
     @abc.abstractmethod
     async def delete_collection(
         self, collection_id: str, **kwargs
diff --git a/stac_fastapi/types/stac_fastapi/types/stac.py b/stac_fastapi/types/stac_fastapi/types/stac.py
index 7cfdb3d..dfd9a62 100644
--- a/stac_fastapi/types/stac_fastapi/types/stac.py
+++ b/stac_fastapi/types/stac_fastapi/types/stac.py
@@ -1,8 +1,10 @@
 """STAC types."""
 
-from typing import Any, Dict, List, Literal, Union
+import json
+from typing import Any, Dict, List, Literal, Optional, Union
 
-from stac_pydantic.shared import BBox
+from pydantic import ConfigDict, Field
+from stac_pydantic.shared import BBox, StacBaseModel
 from typing_extensions import NotRequired, TypedDict
 
 NumType = Union[float, int]
@@ -77,3 +79,152 @@ class Collections(TypedDict):
     links: List[Dict[str, Any]]
     numberMatched: NotRequired[int]
     numberReturned: NotRequired[int]
+
+
+class PatchAddReplaceTest(StacBaseModel):
+    """Add, Replace or Test Operation."""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {"op": "add", "path": "/properties/foo", "value": "bar"},
+                {"op": "replace", "path": "/properties/foo", "value": "bar"},
+                {"op": "test", "path": "/properties/foo", "value": "bar"},
+            ]
+        }
+    )
+
+    path: str
+    op: Literal["add", "replace", "test"]
+    value: Any
+
+    @property
+    def json_value(self) -> str:
+        """JSON dump of value field.
+
+        Returns:
+            str: JSON-ised value
+        """
+        return json.dumps(self.value)
+
+
+class PatchRemove(StacBaseModel):
+    """Remove Operation."""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "op": "remove",
+                    "path": "/properties/foo",
+                }
+            ]
+        }
+    )
+
+    path: str
+    op: Literal["remove"]
+
+
+class PatchMoveCopy(StacBaseModel):
+    """Move or Copy Operation."""
+
+    model_config = ConfigDict(
+        populate_by_name=True,
+        json_schema_extra={
+            "examples": [
+                {
+                    "op": "copy",
+                    "path": "/properties/foo",
+                    "from": "/properties/bar",
+                },
+                {
+                    "op": "move",
+                    "path": "/properties/foo",
+                    "from": "/properties/bar",
+                },
+            ]
+        },
+    )
+
+    path: str
+    op: Literal["move", "copy"]
+    from_: str = Field(alias="from")
+
+
+PatchOperation = Union[PatchAddReplaceTest, PatchMoveCopy, PatchRemove]
+
+
+class BasePartial(StacBaseModel):
+    """Base Partial Class."""
+
+    @staticmethod
+    def merge_to_operations(data: Dict) -> List[PatchOperation]:
+        """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(PatchRemove(op="remove", path=f"/{key}"))
+
+            elif isinstance(value, dict):
+                nested_operations = BasePartial.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(
+                    PatchAddReplaceTest(op="add", path=f"/{key}", value=value)
+                )
+
+        return operations
+
+    def operations(self) -> List[PatchOperation]:
+        """Equivalent RF6902 operations to merge of Partial.
+
+        Returns:
+            List[PatchOperation]: Equivalent list of RF6902 operations
+        """
+        return self.merge_to_operations(self.model_dump())
+
+
+class PartialCollection(BasePartial):
+    """Partial STAC Collection."""
+
+    type: Optional[str] = None
+    stac_version: Optional[str] = None
+    stac_extensions: Optional[List[str]] = None
+    id: Optional[str] = None
+    title: Optional[str] = None
+    description: Optional[str] = None
+    links: Optional[Dict[str, Any]] = None
+    keywords: Optional[List[str]] = None
+    license: Optional[str] = None
+    providers: Optional[List[Dict[str, Any]]] = None
+    extent: Optional[Dict[str, Any]] = None
+    summaries: Optional[Dict[str, Any]] = None
+    assets: Optional[Dict[str, Any]] = None
+
+
+class PartialItem(BasePartial):
+    """Partial STAC Item."""
+
+    type: Optional[Literal["Feature"]] = None
+    stac_version: Optional[str] = None
+    stac_extensions: Optional[List[str]] = None
+    id: Optional[str] = None
+    geometry: Optional[Dict[str, Any]] = None
+    bbox: Optional[BBox] = None
+    properties: Optional[Dict[str, Any]] = None
+    links: Optional[List[Dict[str, Any]]] = None
+    assets: Optional[Dict[str, Any]] = None
+    collection: Optional[str] = None


Test patch:
diff --git a/stac_fastapi/api/tests/test_api.py b/stac_fastapi/api/tests/test_api.py
index 7db4d9a..d4ebf95 100644
--- a/stac_fastapi/api/tests/test_api.py
+++ b/stac_fastapi/api/tests/test_api.py
@@ -3,10 +3,7 @@ from starlette.testclient import TestClient
 
 from stac_fastapi.api.app import StacApi
 from stac_fastapi.api.models import ItemCollectionUri, create_request_model
-from stac_fastapi.extensions.core import (
-    TokenPaginationExtension,
-    TransactionExtension,
-)
+from stac_fastapi.extensions.core import TokenPaginationExtension, TransactionExtension
 from stac_fastapi.types import config, core
 
 
@@ -430,6 +427,9 @@ class DummyTransactionsClient(core.BaseTransactionsClient):
     def update_item(self, *args, **kwargs):
         return "dummy response"
 
+    def patch_item(self, *args, **kwargs):
+        return "dummy response"
+
     def delete_item(self, *args, **kwargs):
         return "dummy response"
 
@@ -439,6 +439,9 @@ class DummyTransactionsClient(core.BaseTransactionsClient):
     def update_collection(self, *args, **kwargs):
         return "dummy response"
 
+    def patch_collection(self, *args, **kwargs):
+        return "dummy response"
+
     def delete_collection(self, *args, **kwargs):
         return "dummy response"
 
diff --git a/stac_fastapi/extensions/tests/test_transaction.py b/stac_fastapi/extensions/tests/test_transaction.py
index 689e519..2f6487b 100644
--- a/stac_fastapi/extensions/tests/test_transaction.py
+++ b/stac_fastapi/extensions/tests/test_transaction.py
@@ -1,5 +1,5 @@
 import json
-from typing import Iterator, Union
+from typing import Iterator, List, Union
 
 import pytest
 from stac_pydantic import Collection
@@ -11,6 +11,7 @@ from stac_fastapi.api.app import StacApi
 from stac_fastapi.extensions.core import TransactionExtension
 from stac_fastapi.types.config import ApiSettings
 from stac_fastapi.types.core import BaseCoreClient, BaseTransactionsClient
+from stac_fastapi.types.stac import PartialCollection, PartialItem, PatchOperation
 
 
 class DummyCoreClient(BaseCoreClient):
@@ -46,6 +47,22 @@ class DummyTransactionsClient(BaseTransactionsClient):
             "type": item.type,
         }
 
+    def patch_item(
+        self,
+        collection_id: str,
+        item_id: str,
+        patch: Union[PartialItem, List[PatchOperation]],
+        **kwargs,
+    ):
+        if isinstance(patch, PartialItem):
+            patch = patch.operations()
+
+        return {
+            "path_collection_id": collection_id,
+            "path_item_id": item_id,
+            "patch": patch,
+        }
+
     def delete_item(self, item_id: str, collection_id: str, **kwargs):
         return {
             "path_collection_id": collection_id,
@@ -58,6 +75,20 @@ class DummyTransactionsClient(BaseTransactionsClient):
     def update_collection(self, collection_id: str, collection: Collection, **kwargs):
         return {"path_collection_id": collection_id, "type": collection.type}
 
+    def patch_collection(
+        self,
+        collection_id: str,
+        patch: Union[PartialCollection, List[PatchOperation]],
+        **kwargs,
+    ):
+        if isinstance(patch, PartialCollection):
+            patch = patch.operations()
+
+        return {
+            "path_collection_id": collection_id,
+            "patch": patch,
+        }
+
     def delete_collection(self, collection_id: str, **kwargs):
         return {"path_collection_id": collection_id}
 
@@ -88,6 +119,33 @@ def test_update_item(client: TestClient, item: Item) -> None:
     assert response.json()["type"] == "Feature"
 
 
+def test_patch_operation_item(client: TestClient) -> None:
+    response = client.patch(
+        "/collections/a-collection/items/an-item",
+        content='[{"op": "add", "path": "/properties/foo", "value": "bar"}]',
+    )
+    assert response.is_success, response.text
+    assert response.json()["path_collection_id"] == "a-collection"
+    assert response.json()["path_item_id"] == "an-item"
+    assert response.json()["patch"] == [
+        {"op": "add", "path": "/properties/foo", "value": "bar"}
+    ]
+
+
+def test_patch_merge_item(client: TestClient) -> None:
+    response = client.patch(
+        "/collections/a-collection/items/an-item",
+        content=json.dumps({"properties": {"hello": "world", "foo": None}}),
+    )
+    assert response.is_success, response.text
+    assert response.json()["path_collection_id"] == "a-collection"
+    assert response.json()["path_item_id"] == "an-item"
+    assert response.json()["patch"] == [
+        {"op": "add", "path": "/properties/hello", "value": "world"},
+        {"op": "remove", "path": "/properties/foo"},
+    ]
+
+
 def test_delete_item(client: TestClient) -> None:
     response = client.delete("/collections/a-collection/items/an-item")
     assert response.is_success, response.text
@@ -108,6 +166,31 @@ def test_update_collection(client: TestClient, collection: Collection) -> None:
     assert response.json()["type"] == "Collection"
 
 
+def test_patch_operation_collection(client: TestClient) -> None:
+    response = client.patch(
+        "/collections/a-collection",
+        content='[{"op": "add", "path": "/properties/foo", "value": "bar"}]',
+    )
+    assert response.is_success, response.text
+    assert response.json()["path_collection_id"] == "a-collection"
+    assert response.json()["patch"] == [
+        {"op": "add", "path": "/properties/foo", "value": "bar"}
+    ]
+
+
+def test_patch_merge_collection(client: TestClient) -> None:
+    response = client.patch(
+        "/collections/a-collection",
+        content=json.dumps({"summaries": {"hello": "world", "foo": None}}),
+    )
+    assert response.is_success, response.text
+    assert response.json()["path_collection_id"] == "a-collection"
+    assert response.json()["patch"] == [
+        {"op": "add", "path": "/summaries/hello", "value": "world"},
+        {"op": "remove", "path": "/summaries/foo"},
+    ]
+
+
 def test_delete_collection(client: TestClient, collection: Collection) -> None:
     response = client.delete("/collections/a-collection")
     assert response.is_success, response.text