[Cosmos] Patch API implementation (Azure#29497)

* Patch API implementation with samples, tests, changelog

* Update documents.py

* Update CHANGELOG.md

* use StatusCodes instead of direct ints

* mark patch as provisional (preview)

* update version, typehints, pylint

* remove PatchOperationType class

* Update __init__.py

* fix typehints

* Update CHANGELOG.md

Author: simorenoh

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -1,8 +1,9 @@
 ## Release History
 
-### 4.3.2 (Unreleased)
+### 4.4.0b1 (Unreleased)
 
 #### Features Added
+- Added **preview** partial document update (Patch API) functionality and container methods for patching items with operations. See [PR 29497](https://github.com/Azure/azure-sdk-for-python/pull/29497). For more information on Patch, please see [Azure Cosmos DB Partial Document Update](https://learn.microsoft.com/azure/cosmos-db/partial-document-update).
 
 #### Breaking Changes
 

sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py

@@ -1731,6 +1731,36 @@ def ReplaceItem(self, document_link, new_document, options=None, **kwargs):
 
         return self.Replace(new_document, path, "docs", document_id, None, options, **kwargs)
 
+    def PatchItem(self, document_link, operations, options=None, **kwargs):
+        """Patches a document and returns it.
+
+        :param str document_link: The link to the document.
+        :param list operations: The operations for the patch request.
+        :param dict options: The request options for the request.
+
+        :return:
+            The new Document.
+        :rtype:
+            dict
+
+        """
+        path = base.GetPathFromLink(document_link)
+        document_id = base.GetResourceIdOrFullNameFromLink(document_link)
+        typ = "docs"
+
+        if options is None:
+            options = {}
+
+        initial_headers = self.default_headers
+        headers = base.GetHeaders(self, initial_headers, "patch", path, document_id, typ, options)
+        # Patch will use WriteEndpoint since it uses PUT operation
+        request_params = _request_object.RequestObject(typ, documents._OperationType.Patch)
+        result, self.last_response_headers = self.__Patch(path, request_params, operations, headers, **kwargs)
+
+        # update session for request mutates data on server side
+        self._UpdateSessionIfRequired(headers, result, self.last_response_headers)
+        return result
+
     def DeleteItem(self, document_link, options=None, **kwargs):
         """Deletes a document.
 
@@ -2309,6 +2339,32 @@ def __Put(self, path, request_params, body, req_headers, **kwargs):
             **kwargs
         )
 
+    def __Patch(self, path, request_params, operations, req_headers, **kwargs):
+        """Azure Cosmos 'PATCH' http request.
+
+        :params str path:
+        :params ~azure.cosmos.RequestObject request_params:
+        :params list operations:
+        :params dict req_headers:
+
+        :return:
+            Tuple of (result, headers).
+        :rtype:
+            tuple of (dict, dict)
+
+        """
+        request = self.pipeline_client.patch(url=path, headers=req_headers)
+        return synchronized_request.SynchronizedRequest(
+            client=self,
+            request_params=request_params,
+            global_endpoint_manager=self._global_endpoint_manager,
+            connection_policy=self.connection_policy,
+            pipeline_client=self.pipeline_client,
+            request=request,
+            request_data=operations,
+            **kwargs
+        )
+
     def __Delete(self, path, request_params, req_headers, **kwargs):
         """Azure Cosmos 'DELETE' http request.
 

sdk/cosmos/azure-cosmos/azure/cosmos/_version.py

@@ -19,4 +19,4 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-VERSION = "4.3.2"
+VERSION = "4.4.0b1"

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_container.py

@@ -22,7 +22,7 @@
 """Create, read, update and delete items in the Azure Cosmos DB SQL API service.
 """
 
-from typing import Any, Dict, Optional, Union, cast, Awaitable
+from typing import Any, Dict, Optional, Union, cast, Awaitable, List
 from azure.core.async_paging import AsyncItemPaged
 
 from azure.core.tracing.decorator import distributed_trace
@@ -522,6 +522,50 @@ async def replace_item(
             response_hook(self.client_connection.last_response_headers, result)
         return result
 
+    @distributed_trace_async
+    async def patch_item(
+        self,
+        item: Union[str, Dict[str, Any]],
+        partition_key: Union[str, int, float, bool],
+        patch_operations: List[Dict[str, Any]],
+        **kwargs: Any
+    ) -> Dict[str, Any]:
+        """ **Provisional method** Patches the specified item with the provided operations if it
+         exists in the container.
+
+        If the item does not already exist in the container, an exception is raised.
+
+        :param item: The ID (name) or dict representing item to be patched.
+        :type item: Union[str, Dict[str, Any]]
+        :param partition_key: The partition key of the object to patch.
+        :type partition_key: Union[str, int, float, bool]
+        :param patch_operations: The list of patch operations to apply to the item.
+        :type patch_operations: List[Dict[str, Any]]
+        :keyword str pre_trigger_include: trigger id to be used as pre operation trigger.
+        :keyword str post_trigger_include: trigger id to be used as post operation trigger.
+        :keyword str session_token: Token for use with Session consistency.
+        :keyword dict[str,str] initial_headers: Initial headers to be sent as part of the request.
+        :keyword str etag: An ETag value, or the wildcard character (*). Used to check if the resource
+            has changed, and act according to the condition specified by the `match_condition` parameter.
+        :keyword ~azure.core.MatchConditions match_condition: The match condition to use upon the etag.
+        :keyword Callable response_hook: A callable invoked with the response metadata.
+        :returns: A dict representing the item after the patch operations went through.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: The patch operations failed or the item with
+            given id does not exist.
+        :rtype: dict[str, Any]
+        """
+        request_options = _build_options(kwargs)
+        response_hook = kwargs.pop('response_hook', None)
+        request_options["disableAutomaticIdGeneration"] = True
+        request_options["partitionKey"] = partition_key
+
+        item_link = self._get_document_link(item)
+        result = await self.client_connection.PatchItem(
+            document_link=item_link, operations=patch_operations, options=request_options, **kwargs)
+        if response_hook:
+            response_hook(self.client_connection.last_response_headers, result)
+        return result
+
     @distributed_trace_async
     async def delete_item(
         self,

sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py

@@ -1160,6 +1160,36 @@ async def ReplaceItem(self, document_link, new_document, options=None, **kwargs)
 
         return await self.Replace(new_document, path, "docs", document_id, None, options, **kwargs)
 
+    async def PatchItem(self, document_link, operations, options=None, **kwargs):
+        """Patches a document and returns it.
+
+        :param str document_link: The link to the document.
+        :param list operations: The operations for the patch request.
+        :param dict options: The request options for the request.
+
+        :return:
+            The new Document.
+        :rtype:
+            dict
+
+        """
+        path = base.GetPathFromLink(document_link)
+        document_id = base.GetResourceIdOrFullNameFromLink(document_link)
+        typ = "docs"
+
+        if options is None:
+            options = {}
+
+        initial_headers = self.default_headers
+        headers = base.GetHeaders(self, initial_headers, "patch", path, document_id, typ, options)
+        # Patch will use WriteEndpoint since it uses PUT operation
+        request_params = _request_object.RequestObject(typ, documents._OperationType.Patch)
+        result, self.last_response_headers = await self.__Patch(path, request_params, operations, headers, **kwargs)
+
+        # update session for request mutates data on server side
+        self._UpdateSessionIfRequired(headers, result, self.last_response_headers)
+        return result
+
     async def ReplaceOffer(self, offer_link, offer, **kwargs):
         """Replaces an offer and returns it.
 
@@ -1263,6 +1293,32 @@ async def __Put(self, path, request_params, body, req_headers, **kwargs):
             **kwargs
         )
 
+    async def __Patch(self, path, request_params, operations, req_headers, **kwargs):
+        """Azure Cosmos 'PATCH' http request.
+
+        :params str path:
+        :params ~azure.cosmos.RequestObject request_params:
+        :params list operations:
+        :params dict req_headers:
+
+        :return:
+            Tuple of (result, headers).
+        :rtype:
+            tuple of (dict, dict)
+
+        """
+        request = self.pipeline_client.patch(url=path, headers=req_headers)
+        return await asynchronous_request.AsynchronousRequest(
+            client=self,
+            request_params=request_params,
+            global_endpoint_manager=self._global_endpoint_manager,
+            connection_policy=self.connection_policy,
+            pipeline_client=self.pipeline_client,
+            request=request,
+            request_data=operations,
+            **kwargs
+        )
+
     async def DeleteDatabase(self, database_link, options=None, **kwargs):
         """Deletes a database.
 

sdk/cosmos/azure-cosmos/azure/cosmos/container.py

@@ -570,6 +570,50 @@ def create_item(
             response_hook(self.client_connection.last_response_headers, result)
         return result
 
+    @distributed_trace
+    def patch_item(
+        self,
+        item: Union[str, Dict[str, Any]],
+        partition_key: Union[str, int, float, bool],
+        patch_operations: List[Dict[str, Any]],
+        **kwargs:Any
+    ) -> Dict[str, Any]:
+        """ **Provisional method** Patches the specified item with the provided operations if it
+         exists in the container.
+
+        If the item does not already exist in the container, an exception is raised.
+
+        :param item: The ID (name) or dict representing item to be patched.
+        :type item: Union[str, Dict[str, Any]]
+        :param partition_key: The partition key of the object to patch.
+        :type partition_key: Union[str, int, float, bool]
+        :param patch_operations: The list of patch operations to apply to the item.
+        :type patch_operations: List[Dict[str, Any]]
+        :keyword str pre_trigger_include: trigger id to be used as pre operation trigger.
+        :keyword str post_trigger_include: trigger id to be used as post operation trigger.
+        :keyword str session_token: Token for use with Session consistency.
+        :keyword dict[str,str] initial_headers: Initial headers to be sent as part of the request.
+        :keyword str etag: An ETag value, or the wildcard character (*). Used to check if the resource
+            has changed, and act according to the condition specified by the `match_condition` parameter.
+        :keyword ~azure.core.MatchConditions match_condition: The match condition to use upon the etag.
+        :keyword Callable response_hook: A callable invoked with the response metadata.
+        :returns: A dict representing the item after the patch operations went through.
+        :raises ~azure.cosmos.exceptions.CosmosHttpResponseError: The patch operations failed or the item with
+            given id does not exist.
+        :rtype: dict[str, Any]
+        """
+        request_options = build_options(kwargs)
+        response_hook = kwargs.pop('response_hook', None)
+        request_options["disableAutomaticIdGeneration"] = True
+        request_options["partitionKey"] = partition_key
+
+        item_link = self._get_document_link(item)
+        result = self.client_connection.PatchItem(
+            document_link=item_link, operations=patch_operations, options=request_options, **kwargs)
+        if response_hook:
+            response_hook(self.client_connection.last_response_headers, result)
+        return result
+
     @distributed_trace
     def delete_item(
         self,

sdk/cosmos/azure-cosmos/azure/cosmos/documents.py

@@ -381,13 +381,14 @@ class _OperationType(object):
     ExecuteJavaScript = "ExecuteJavaScript"
     Head = "Head"
     HeadFeed = "HeadFeed"
+    Patch = "Patch"
     Query = "Query"
+    QueryPlan = "QueryPlan"
     Read = "Read"
     ReadFeed = "ReadFeed"
     Recreate = "Recreate"
     Replace = "Replace"
     SqlQuery = "SqlQuery"
-    QueryPlan = "QueryPlan"
     Update = "Update"
     Upsert = "Upsert"
 

sdk/cosmos/azure-cosmos/samples/document_management.py

@@ -102,9 +102,27 @@ def upsert_item(container, doc_id):
 
     print('Upserted Item\'s Id is {0}, new subtotal={1}'.format(response['id'], response['subtotal']))
 
+def patch_item(container, doc_id):
+    print('\n1.7 Patching Item by Id\n')
+
+    operations = [
+        {"op": "add", "path": "/favorite_color", "value": "red"},
+        {"op": "remove", "path": "/ttl"},
+        {"op": "replace", "path": "/tax_amount", "value": 14},
+        {"op": "set", "path": "/items/0/discount", "value": 20.0512},
+        {"op": "incr", "path": "/total_due", "value": 5},
+        {"op": "move", "from": "/freight", "path": "/service_addition"}
+    ]
+
+    response = container.patch_item(item=doc_id, partition_key=doc_id, patch_operations=operations)
+    print('Patched Item\'s Id is {0}, new path favorite color={1}, removed path ttl={2}, replaced path tax_amount={3},'
+          ' set path for item at index 0 of discount={4}, increase in path total_due, new total_due={5}, move from path freight={6}'
+          ' to path service_addition={7}'.format(response["id"], response["favorite_color"], response.get("ttl"),
+                                                 response["tax_amount"], response["items"].get(0).get("discount"),
+                                                 response["total_due"], response.get("freight"), response["service_addition"]))
 
 def delete_item(container, doc_id):
-    print('\n1.7 Deleting Item by Id\n')
+    print('\n1.8 Deleting Item by Id\n')
 
     response = container.delete_item(item=doc_id, partition_key=doc_id)
 
@@ -175,6 +193,7 @@ def run_sample():
         query_items(container, 'SalesOrder1')
         replace_item(container, 'SalesOrder1')
         upsert_item(container, 'SalesOrder1')
+        patch_item(container, 'SalesOrder1')
         delete_item(container, 'SalesOrder1')
 
         # cleanup database after sample

sdk/cosmos/azure-cosmos/samples/document_management_async.py

@@ -120,9 +120,27 @@ async def upsert_item(container, doc_id):
 
     print('Upserted Item\'s Id is {0}, new subtotal={1}'.format(response['id'], response['subtotal']))
 
+async def patch_item(container, doc_id):
+    print('\n1.7 Patching Item by Id\n')
+
+    operations = [
+        {"op": "add", "path": "/favorite_color", "value": "red"},
+        {"op": "remove", "path": "/ttl"},
+        {"op": "replace", "path": "/tax_amount", "value": 14},
+        {"op": "set", "path": "/items/0/discount", "value": 20.0512},
+        {"op": "incr", "path": "/total_due", "value": 5},
+        {"op": "move", "from": "/freight", "path": "/service_addition"}
+    ]
+
+    response = await container.patch_item(item=doc_id, partition_key=doc_id, patch_operations=operations)
+    print('Patched Item\'s Id is {0}, new path favorite color={1}, removed path ttl={2}, replaced path tax_amount={3},'
+          ' set path for item at index 0 of discount={4}, increase in path total_due, new total_due={5}, '
+          'move from path freight={6} to path service_addition={7}'.format(response["id"], response["favorite_color"],
+                                response.get("ttl"), response["tax_amount"], response["items"].get(0).get("discount"),
+                                response["total_due"], response.get("freight"), response["service_addition"]))
 
 async def delete_item(container, doc_id):
-    print('\n1.7 Deleting Item by Id\n')
+    print('\n1.8 Deleting Item by Id\n')
 
     await container.delete_item(item=doc_id, partition_key=doc_id)
 
@@ -195,6 +213,7 @@ async def run_sample():
             await query_items(container, 'SalesOrder1')
             await replace_item(container, 'SalesOrder1')
             await upsert_item(container, 'SalesOrder1')
+            await patch_item(container, 'SalesOrder1')
             await delete_item(container, 'SalesOrder1')
 
             # cleanup database after sample