如何用Python清空Azure CosmosDB容器？遇CosmosHttpResponseError报错

一、批量删除文档 #

适合需要保留容器配置的场景，通过查询+批量操作实现：

from azure.cosmos import CosmosClient
import asyncio

async def bulk_delete_all_items(client, database_id, container_id):
    database = client.get_database_client(database_id)
    container = database.get_container_client(container_id)
    
    # 替换为你的分区键字段名，跨分区查询需开启enable_cross_partition_query
    query = "SELECT c.id, c.your_partition_key FROM c"
    items = []
    continuation_token = None
    
    # 分页查询避免内存溢出
    while True:
        results = container.query_items(
            query=query,
            enable_cross_partition_query=True,
            max_item_count=1000,
            continuation_token=continuation_token
        )
        batch = list(results)
        if not batch:
            break
        items.extend(batch)
        continuation_token = results.continuation_token
    
    # 批量构建删除操作
    operations = [("delete", item) for item in items]
    await container.execute_bulk_async(operations)

# 初始化客户端（替换为你的账户信息）
client = CosmosClient("your_account_endpoint", "your_account_key")
asyncio.run(bulk_delete_all_items(client, "your_db_id", "your_container_id"))

二、删除并重建容器 #

数据量极大时最高效，直接重置容器：

from azure.cosmos import CosmosClient, PartitionKey

client = CosmosClient("your_account_endpoint", "your_account_key")
database = client.get_database_client("your_db_id")
container_id = "your_container_id"

# 删除原容器
database.delete_container(container_id)
# 重新创建容器（替换为你的分区键和配置）
database.create_container(
    id=container_id,
    partition_key=PartitionKey(path="/your_partition_key"),
    offer_throughput=400  # 按需设置吞吐量
)

三、使用存储过程批量删除 #

适合在Cosmos DB端执行操作，减少网络开销：

存储过程代码（JavaScript） #

function deleteAllItems() {
    const collection = getContext().getCollection();
    const query = 'SELECT c._self FROM c';
    let continuationToken = null;

    function deleteBatch(items) {
        if (items.length === 0) {
            getContext().getResponse().setBody("All items deleted");
            return;
        }
        const itemLink = items[0]._self;
        collection.deleteDocument(itemLink, (err) => {
            if (err) throw err;
            deleteBatch(items.slice(1));
        });
    }

    function queryNext() {
        const requestOptions = { continuation: continuationToken };
        collection.queryDocuments(collection.getSelfLink(), query, requestOptions, (err, results, responseOpts) => {
            if (err) throw err;
            continuationToken = responseOpts.continuation;
            deleteBatch(results);
            if (continuationToken) queryNext();
        });
    }

    queryNext();
}

Python调用存储过程 #

from azure.cosmos import CosmosClient

client = CosmosClient("your_account_endpoint", "your_account_key")
container = client.get_database_client("your_db_id").get_container_client("your_container_id")
# 假设已将上述存储过程上传并命名为deleteAllItems
container.execute_stored_procedure("deleteAllItems")

四、CosmosHttpResponseError错误排查 #

• 权限验证：确认账户密钥/资源令牌拥有Write或更高权限，避免只读权限导致删除失败。
• 分区键问题：分区容器操作需指定正确的分区键，跨分区查询/操作必须开启enable_cross_partition_query=True。
• 速率限制处理：遇到429错误时，配置SDK重试策略：

  from azure.cosmos import CosmosClient, ConnectionPolicy, RetryOptions
  
  retry_opts = RetryOptions(max_retry_attempts=15, max_retry_wait_time_in_seconds=45)
  conn_policy = ConnectionPolicy()
  conn_policy.RetryOptions = retry_opts
  client = CosmosClient("your_endpoint", "your_key", connection_policy=conn_policy)
  

• 端点/密钥正确性：检查账户端点是否包含https://前缀，密钥是否存在拼写错误。

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