使用Python SDK执行Azure CosmosDB批量操作时Replace报错求助

在使用Azure CosmosDB Python SDK执行批量操作时，仅Replace操作失败，Create、Delete、Read、Upsert操作均可正常执行。以下是问题详情及解决方案：

问题重现代码 #

import azure.cosmos.cosmos_client as cosmos_client
import azure.cosmos.exceptions as exceptions
from azure.cosmos.http_constants import StatusCodes
from azure.cosmos.partition_key import PartitionKey
import datetime

import config

HOST = config.settings['host']
MASTER_KEY = config.settings['master_key']
DATABASE_ID = config.settings['database_id']
CONTAINER_ID = config.settings['container_id']
CONTAINER_MH_ID = config.settings['container_mh_id']

def execute_item_batch(database):
    print('\n1.10 Executing Batch Item operations\n')
    container = database.create_container_if_not_exists(id="batch_container",
                                                        partition_key=PartitionKey(path='/account_number'))
    # 创建测试用的三个项
    container.create_item(get_sales_order("read_item"))
    container.create_item(get_sales_order("delete_item"))
    container.create_item(get_sales_order("replace_item"))


    # 定义批量操作
    create_item_operation = ("create", (get_sales_order("create_item"),))
    upsert_item_operation = ("upsert", (get_sales_order("upsert_item"),))
    read_item_operation = ("read", ("read_item",))
    delete_item_operation = ("delete", ("delete_item",))
    replace_item_operation = ("replace", ("replace_item", {"id": "replace_item", "message": "item was replaced"}))
    # replace_item_if_match_operation = ("replace",
    #                                    ("replace_item", {"id": "replace_item", "message": "item was replaced"}),
    #                                    {"if_match_etag": container.client_connection.last_response_headers.get("etag")})
    # replace_item_if_none_match_operation = ("replace",
    #                                        ("replace_item", {"id": "replace_item", "message": "item was replaced"}),
    #                                        {"if_none_match_etag":
    #                                             container.client_connection.last_response_headers.get("etag")})

    # 组装批量操作列表
    batch_operations = [
        create_item_operation,
        upsert_item_operation,
        read_item_operation,
        delete_item_operation,
        replace_item_operation,
        # replace_item_if_match_operation,
        # replace_item_if_none_match_operation
        ]

    # 执行批量操作
    batch_results = container.execute_item_batch(batch_operations=batch_operations, partition_key='Account1')
    # 批量操作结果会以列表返回，若其中任一操作失败则抛出CosmosBatchOperationError
    print("\n批量操作结果: {}\n".format(batch_results))

def run_sample():
    client = cosmos_client.CosmosClient(HOST, {'masterKey': MASTER_KEY})
    try:
        # 创建/获取数据库
        db = client.create_database_if_not_exists(id=DATABASE_ID)
        # 执行批量操作示例
        execute_item_batch(db)

    finally:
        print("\nrun_sample 执行完成")

if __name__ == '__main__':
    run_sample()

数据库项截图 #

错误详情 #

Traceback (most recent call last):
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\document_management.py", line 552, in <module>
    run_sample()
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\document_management.py", line 518, in run_sample
    execute_item_batch(db)
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\document_management.py", line 225, in execute_item_batch
    batch_results = container.execute_item_batch(batch_operations=batch_operations, partition_key='Account1')
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\lib\site-packages\azure\core\tracing\decorator.py", line 78, in wrapper_use_tracer
    return func(*args, **kwargs)
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\lib\site-packages\azure\cosmos\container.py", line 709, in execute_item_batch
    result = self.client_connection.Batch(
  File "C:xxxxxxxxxxxxxxxxxxxxcosmosdb\lib\site-packages\azure\cosmos\_cosmos_client_connection.py", line 1784, in Batch
    raise exceptions.CosmosBatchOperationError(error_index=error_index,
azure.cosmos.exceptions.CosmosBatchOperationError: Status code: 400 Sub-status: 1001
There was an error in the transactional batch on index 0. Error message: BAD_REQUEST - Request being sent is invalid.

问题排查 #

1. Replace操作的Item结构不完整：当前代码中Replace操作传入的替换对象仅包含id和message字段，缺少容器定义的分区键字段account_number。CosmosDB要求替换操作必须提供完整的合法Item，分区键是必填项，缺失会触发400 Bad Request错误。
2. 事务批量的原子性特性：CosmosDB事务批量操作是原子性的，只要其中一个操作失败，整个批量会回滚，错误提示的索引0是批量执行时第一个抛出错误的位置，但实际根源是Replace操作的参数不合法。

修复建议 #

方案1：补充分区键字段到替换对象 #

修改Replace操作的定义，确保替换Item包含分区键account_number：

# 修正replace_item_operation，添加分区键字段
replace_item_operation = ("replace", ("replace_item", {"id": "replace_item", "message": "item was replaced", "account_number": "Account1"}))

方案2：基于原Item修改后替换（更严谨） #

先读取目标Item，修改指定字段后再执行替换，确保Item结构完整：

# 先读取原Item，确保包含所有必要字段
original_item = container.read_item(item="replace_item", partition_key="Account1")
original_item["message"] = "item was replaced"
# 使用完整的Item对象执行替换
replace_item_operation = ("replace", original_item)

额外验证 #

确保get_sales_order函数生成的Item包含account_number字段，例如：

def get_sales_order(item_id):
    return {
        "id": item_id,
        "account_number": "Account1",
        # 其他业务字段...
    }

内容的提问来源于stack exchange，提问作者Raghu Monavarthi