Azure · Simon Moreno (simorenoh) · Sep 9, 2026 · Sep 5, 2026 · Sep 6, 2026 · Sep 6, 2026
@@ -2,13 +2,18 @@
 > [!IMPORTANT]
 > We strongly recommend that customers use at least version 4.16.3 of `azure-cosmos`.
 
-### 4.16.5 (Unreleased)
+### 4.17.0 (Unreleased)
 
 #### Features Added
+* Added the `enable_compact_utf8_item_writes` client option. Set it to `True` to reduce item write request sizes by
+  serializing valid Unicode as compact UTF-8 for create, upsert, replace, patch, and transactional batch operations.
+  See [PR 48914](https://github.com/Azure/azure-sdk-for-python/pull/48914).
 
 #### Breaking Changes
 
 #### Bugs Fixed
+* Fixed sync and async item PATCH requests to explicitly send the registered `application/json-patch+json`
+  content type. See [PR 48914](https://github.com/Azure/azure-sdk-for-python/pull/48914).
 
 #### Other Changes
 

@@ -81,6 +81,30 @@ KEY = os.environ['ACCOUNT_KEY']
 client = CosmosClient(URL, credential=KEY)
 ```
 
+### Compact UTF-8 item writes
+
+By default, the client escapes non-ASCII characters in JSON item bodies, so each such character is sent as a
+`\uXXXX` escape sequence. A CJK character that occupies 3 bytes as UTF-8 therefore occupies 6 bytes on the wire, and
+a 4-byte emoji occupies 12. For Unicode-heavy items this expansion increases the size of the request sent over the
+network and can push a request past the 2 MiB request size limit. Applications writing such items can opt in to
+compact UTF-8 serialization when creating the client:
+
+```python
+client = CosmosClient(
+    URL,
+    credential=KEY,
+    enable_compact_utf8_item_writes=True,
+)
+```
+
+The option is disabled by default and applies to create, upsert, replace, patch, and transactional batch item
+bodies. Queries, control-plane requests, and responses are unchanged. Semantic request headers - including the
+partition-key header - are also unchanged; body-derived headers such as `Content-Length` necessarily reflect the
+compact byte count and are recalculated accordingly. Both representations describe the
+same JSON document, so the values stored in the service and returned on reads are identical - only the encoding of
+the outgoing request body differs. See the [synchronous][sample_compact_utf8_item_writes] and
+[asynchronous][sample_compact_utf8_item_writes_async] samples for complete examples.
+
 ### AAD Authentication
 
 You can also authenticate a client utilizing your service principal's AAD credentials and the azure identity package. 
@@ -1268,6 +1292,8 @@ For more extensive documentation on the Cosmos DB service, see the [Azure Cosmos
 [ref_database]: https://aka.ms/azsdk-python-cosmos-ref-database
 [ref_httpfailure]: https://aka.ms/azsdk-python-cosmos-ref-http-failure
 [sample_database_mgmt]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/database_management.py
+[sample_compact_utf8_item_writes]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/compact_utf8_item_writes.py
+[sample_compact_utf8_item_writes_async]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/compact_utf8_item_writes_async.py
 [sample_document_mgmt]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/document_management.py
 [sample_document_mgmt_async]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/document_management_async.py
 [sample_examples_misc]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/cosmos/azure-cosmos/samples/examples.py

@@ -3914,6 +3914,7 @@ namespace azure.cosmos.scripts
                 consistency_level: Optional[str] = None, 
                 availability_strategy: Union[bool, dict[str, Any]] = False, 
                 availability_strategy_executor: Optional[ThreadPoolExecutor] = None, 
+                enable_compact_utf8_item_writes: bool = False, 
                 **kwargs: Any
             ) -> None: ...
 
@@ -4708,6 +4709,7 @@ namespace azure.cosmos.user
                 consistency_level: Optional[str] = None, 
                 availability_strategy: Union[bool, dict[str, Any]] = False, 
                 availability_strategy_executor: Optional[ThreadPoolExecutor] = None, 
+                enable_compact_utf8_item_writes: bool = False, 
                 **kwargs: Any
             ) -> None: ...
 

@@ -1,3 +1,4 @@
-apiMdSha256: 1538a79b2c2da38fb83fd37ccea945314bf2f34218acda3d8f15e2f0268f3040
-parserVersion: 0.3.30
-pythonVersion: 3.13.14
+apiMdSha256: 54ea9d4c0c0f08c381ee7e69a7468ed9c08e5fb55c7af0fc1b9f6d0d6f770a86
+packageVersion: 4.17.0
+parserVersion: 0.3.31
+pythonVersion: 3.10.21
@@ -394,7 +394,12 @@ def GetHeaders(  # pylint: disable=too-many-statements,too-many-branches
             authorization = urllib_quote(authorization, "-_.!~*'()")
         headers[http_constants.HttpHeaders.Authorization] = authorization
 
-    if verb in ("post", "put"):
+    if verb == "patch":
+        if not headers.get(http_constants.HttpHeaders.ContentType):
+            headers[http_constants.HttpHeaders.ContentType] = (
+                _runtime_constants.MediaTypes.JsonPatch
+            )
+    elif verb in ("post", "put"):
         if not headers.get(http_constants.HttpHeaders.ContentType):
             headers[http_constants.HttpHeaders.ContentType] = _runtime_constants.MediaTypes.Json
 

@@ -134,14 +134,15 @@ class _QueryCompatibilityMode:
     _DefaultStringHashPrecision = 3
     _DefaultStringRangePrecision = -1
 
-    def __init__( # pylint: disable=too-many-statements
+    def __init__( # pylint: disable=too-many-statements, too-many-locals
         self,
         url_connection: str,
         auth: CredentialDict,
         connection_policy: Optional[ConnectionPolicy] = None,
         consistency_level: Optional[str] = None,
         availability_strategy: Union[bool, dict[str, Any]] = False,
         availability_strategy_executor: Optional[ThreadPoolExecutor] = None,
+        enable_compact_utf8_item_writes: bool = False,
         **kwargs: Any
     ) -> None:
         """
@@ -160,10 +161,15 @@ def __init__( # pylint: disable=too-many-statements
             The availability strategy configuration for routing requests across regions.
         :param concurrent.futures.ThreadPoolExecutor availability_strategy_executor:
             The thread pool executor for handling availability strategy requests.
+        :param bool enable_compact_utf8_item_writes:
+            Whether item write bodies should use compact UTF-8 serialization.
         :keyword Literal["High", "Low"] priority: Priority based execution allows users to set a priority for the
             client. Once the user has reached their provisioned throughput, low priority requests are throttled
             before high priority requests start getting throttled. Feature must first be enabled at the account level.
         """
+        self._enable_compact_utf8_item_writes = _utils._validate_enable_compact_utf8_item_writes(
+            enable_compact_utf8_item_writes
+        )
         self.client_id = str(uuid.uuid4())
         self.url_connection = url_connection
         self.availability_strategy: Union[CrossRegionHedgingStrategy, None] =\

@@ -35,6 +35,7 @@ class MediaTypes(object):
     ImagePng = "image/png"
     JavaScript = "application/x-javascript"
     Json = "application/json"
+    JsonPatch = "application/json-patch+json"
     OctetStream = "application/octet-stream"
     QueryJson = "application/query+json"
     SQL = "application/sql"

@@ -21,6 +21,7 @@
 
 """Synchronized request in the Azure Cosmos database service.
 """
+# cspell:ignore surrogatepass
 import copy
 import json
 import time
@@ -37,6 +38,17 @@
 from ._request_object import RequestObject
 from .documents import _OperationType
 
+_ITEM_BODY_WRITE_OPERATIONS = frozenset((
+    _OperationType.Create,
+    _OperationType.Upsert,
+    _OperationType.Replace,
+    _OperationType.Patch,
+    # Batch membership is necessary but not sufficient: a batch made up only of
+    # read and delete operations carries no item body. _should_escape_non_ascii_
+    # in_request_body refines this entry with _batch_contains_item_body below.
+    _OperationType.Batch,
+))
+
 # cspell:ignore ppaf
 def _is_readable_stream(obj):
     """Checks whether obj is a file-like readable stream.
@@ -50,26 +62,104 @@ def _is_readable_stream(obj):
     return False
 
 
-def _request_body_from_data(data):
-    """Gets request body from data.
+def _request_body_from_data(data, ensure_ascii=True):
+    """Convert supported request data into an HTTP body.
+
+    Dictionaries, lists, and tuples are serialized as compact JSON. Other
+    supported body types are returned unchanged.
+
+    When ``ensure_ascii`` is False the serialized body is returned as UTF-8
+    ``bytes``. The escaped default and caller-supplied pre-serialized strings
+    keep returning ``str``, preserving existing behavior.
 
-    When `data` is dict and list into unicode string; otherwise return `data`
-    without making any change.
+    Returning bytes is required for correctness, not just convenience. The sync
+    Requests transport forwards the body to ``requests`` unchanged, and the
+    supported dependency range still permits urllib3 1.x, whose connection path
+    hands ``str`` bodies to ``http.client``. ``http.client`` encodes them as
+    Latin-1, so a compact body would be sent with the wrong encoding: text that
+    has no Latin-1 representation (any CJK or emoji) raises ``UnicodeEncodeError``
+    before the request is sent, and text that does have one is written in Latin-1
+    rather than UTF-8, so 'é' goes out as the single byte 0xE9 instead of
+    0xC3 0xA9 and the service receives different characters than the caller
+    supplied. Handing the transport the encoded bytes removes every subsequent
+    re-encoding step.
 
     :param Union[str, unicode, file-like stream object, dict, list, None] data:
-    :returns: the json dump data.
-    :rtype: Union[str, unicode, file-like stream object, None]
+    :param bool ensure_ascii: Whether non-ASCII characters should be escaped.
+    :returns: The serialized or unchanged request body.
+    :rtype: Union[str, bytes, file-like stream object, None]
 
     """
     if data is None or isinstance(data, str) or _is_readable_stream(data):
         return data
     if isinstance(data, (dict, list, tuple)):
-        json_dumped = json.dumps(data, separators=(",", ":"))
-
-        return json_dumped
+        if ensure_ascii:
+            return json.dumps(data, separators=(",", ":"))
+        json_dumped = json.dumps(data, separators=(",", ":"), ensure_ascii=False)
+        try:
+            # Encode once; callers derive Content-Length directly from these bytes.
+            encoded_body = json_dumped.encode("utf-8")
+        except UnicodeEncodeError:
+            # Rare path for text originating from UTF-16 APIs. Combine adjacent
+            # high/low surrogate pairs into their Unicode scalar while preserving
+            # unpaired surrogates, then escape only those remaining invalid code
+            # units as valid JSON \uXXXX sequences.
+            normalized_body = json_dumped.encode(
+                "utf-16-le", "surrogatepass"
+            ).decode("utf-16-le", "surrogatepass")
+            encoded_body = normalized_body.encode("utf-8", "backslashreplace")
+        # Send exactly these bytes, so no transport re-encodes them.
+        return encoded_body
     return None
 
 
+def _batch_contains_item_body(batch_operations):
+    """Return whether a formatted transactional batch contains an item body.
+
+    Read and delete batch operations contain only an item ID. Create, upsert,
+    replace, and patch operations contain a resourceBody.
+
+    :param list[dict[str, object]] batch_operations: The formatted batch operations.
+    :returns: Whether at least one operation contains an item body.
+    :rtype: bool
+    """
+    return (
+        isinstance(batch_operations, (list, tuple))
+        and any(
+            isinstance(operation, dict) and "resourceBody" in operation
+            for operation in batch_operations
+        )
+    )
+
+
+def _should_escape_non_ascii_in_request_body(client, request_params, request_data):
+    """Decide whether a request body must keep non-ASCII characters escaped.
+
+    Compact UTF-8 is only used when the client opted in and the request is one
+    of the item write operations the option is scoped to. A transactional batch
+    uses compact UTF-8 only when at least one operation contains an item body;
+    read/delete-only batches keep the escaped form. Every other request,
+    including control-plane bodies and queries, also keeps the escaped form.
+
+    :param object client: the client connection issuing the request.
+    :param ~azure.cosmos._request_object.RequestObject request_params: the request parameters.
+    :param object request_data: The body data that will be serialized.
+    :returns: whether non-ASCII characters should be escaped in the body.
+    :rtype: bool
+    """
+    # getattr keeps the safe (escaped) default for any caller that supplies a
+    # client object without the option, e.g. custom or legacy connections.
+    return (
+        not getattr(client, "_enable_compact_utf8_item_writes", False)
+        or request_params.resource_type != http_constants.ResourceType.Document
+        or request_params.operation_type not in _ITEM_BODY_WRITE_OPERATIONS
+        or (
+            request_params.operation_type == _OperationType.Batch
+            and not _batch_contains_item_body(request_data)
+        )
+    )
+
+
 def _Request(global_endpoint_manager, request_params, connection_policy, pipeline_client, request, **kwargs): # pylint: disable=too-many-statements
     """Makes one http request using the requests module.
 
@@ -285,12 +375,21 @@ def SynchronizedRequest(
     :return: tuple of (result, headers)
     :rtype: tuple of (dict dict)
     """
-    request.data = _request_body_from_data(request_data)
-    if request.data and isinstance(request.data, str):
+    request.data = _request_body_from_data(
+        request_data,
+        ensure_ascii=_should_escape_non_ascii_in_request_body(client, request_params, request_data)
+    )
+    if isinstance(request.data, (bytes, bytearray)):
+        # Compact UTF-8 bodies reach the transport as their final bytes, so
+        # Content-Length cannot drift from the wire body.
+        request.headers[http_constants.HttpHeaders.ContentLength] = len(request.data)
+    elif request.data and isinstance(request.data, str):
         # Use UTF-8 byte length, not str length (code-point count), so the
         # header matches the bytes the transport actually writes for any
         # non-ASCII payload.
-        request.headers[http_constants.HttpHeaders.ContentLength] = len(request.data.encode("utf-8"))
+        request.headers[http_constants.HttpHeaders.ContentLength] = len(
+            request.data.encode("utf-8")
+        )
     elif request.data is None:
         request.headers[http_constants.HttpHeaders.ContentLength] = 0
 

@@ -35,6 +35,21 @@
 # cspell:ignore ppcb
 # pylint: disable=protected-access
 
+
+def _validate_enable_compact_utf8_item_writes(value: Any) -> bool:
+    """Validate the compact UTF-8 item-write setting.
+
+    :param value: The setting to validate.
+    :type value: Any
+    :returns: The validated setting.
+    :rtype: bool
+    :raises TypeError: If the setting is not a bool.
+    """
+    if not isinstance(value, bool):
+        raise TypeError("enable_compact_utf8_item_writes must be a bool.")
+    return value
+
+
 def get_user_agent(suffix: Optional[str] = None) -> str:
     os_name = safe_user_agent_header(platform.platform())
     python_version = safe_user_agent_header(platform.python_version())

@@ -19,4 +19,4 @@
 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 # SOFTWARE.
 
-VERSION = "4.16.5"
+VERSION = "4.17.0"
@@ -36,7 +36,11 @@
 from .._constants import _Constants
 from .._request_object import RequestObject
 from .._response_decoding import decode_response_body_for_status
-from .._synchronized_request import _request_body_from_data, _replace_url_prefix
+from .._synchronized_request import (
+    _request_body_from_data,
+    _replace_url_prefix,
+    _should_escape_non_ascii_in_request_body,
+)
 from ..documents import _OperationType
 
 # cspell:ignore ppaf
@@ -238,12 +242,21 @@ async def AsynchronousRequest(
     :return: tuple of (result, headers)
     :rtype: tuple of (dict dict)
     """
-    request.data = _request_body_from_data(request_data)
-    if request.data and isinstance(request.data, str):
+    request.data = _request_body_from_data(
+        request_data,
+        ensure_ascii=_should_escape_non_ascii_in_request_body(client, request_params, request_data)
+    )
+    if isinstance(request.data, (bytes, bytearray)):
+        # Compact UTF-8 bodies reach the transport as their final bytes, so
+        # Content-Length cannot drift from the wire body.
+        request.headers[http_constants.HttpHeaders.ContentLength] = len(request.data)
+    elif request.data and isinstance(request.data, str):
         # Use UTF-8 byte length, not str length (code-point count), so the
         # header matches the bytes the transport actually writes for any
         # non-ASCII payload.
-        request.headers[http_constants.HttpHeaders.ContentLength] = len(request.data.encode("utf-8"))
+        request.headers[http_constants.HttpHeaders.ContentLength] = len(
+            request.data.encode("utf-8")
+        )
     elif request.data is None:
         request.headers[http_constants.HttpHeaders.ContentLength] = 0
 

@@ -192,6 +192,9 @@ class CosmosClient:  # pylint: disable=client-accepts-api-version-keyword
         Default value is False (hedging disabled).
     :paramtype availability_strategy: Union[bool, dict[str, Any]]
     :keyword int availability_strategy_max_concurrency: The max concurrency for parallel requests.
+    :keyword bool enable_compact_utf8_item_writes:
+        Use compact UTF-8 when serializing item bodies for create, upsert, replace, patch, and transactional batch
+        operations. The default is False.
 
     .. admonition:: Example:
 
@@ -216,6 +219,7 @@ def __init__(
     ) -> None:
         """Instantiate a new CosmosClient."""
         auth = _build_auth(credential)
+        enable_compact_utf8_item_writes = kwargs.pop("enable_compact_utf8_item_writes", False)
         connection_policy = _build_connection_policy(kwargs)
         self.client_connection = CosmosClientConnection(
             url_connection=url,
@@ -224,6 +228,7 @@ def __init__(
             connection_policy=connection_policy,
             availability_strategy=availability_strategy,
             availability_strategy_max_concurrency=availability_strategy_max_concurrency,
+            enable_compact_utf8_item_writes=enable_compact_utf8_item_writes,
             **kwargs
         )