如何通过Azure Functions(Python) HTTP触发器覆盖Azure Cosmos DB文档

Cosmos DB的输出绑定默认支持Upsert操作（存在则覆盖，不存在则创建），核心是确保传入的文档包含唯一标识的id字段；若集合为分区模式，还需包含对应分区键字段。

步骤1：调整输出绑定配置（分区集合需操作） #

如果你的Cosmos DB集合设置了分区键，需要在绑定配置中添加partitionKey参数，支持动态从请求中获取值：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "cosmosDB",
      "direction": "out",
      "name": "doc",
      "databaseName": "demodb",
      "collectionName": "data",
      "createIfNotExists": "true",
      "connectionStringSetting": "AzureCosmosDBConnectionString",
      // 分区集合需添加：动态从请求body中获取分区键值
      "partitionKey": "{body.partitionKey}"
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

若集合是非分区模式，可忽略partitionKey配置项。

步骤2：修改函数代码 #

确保请求Body包含id（必选）和分区键（若有），代码示例如下：

import azure.functions as func
import json

def main(req: func.HttpRequest, doc: func.Out[func.Document]) -> func.HttpResponse:
    try:
        # 解析请求JSON体
        request_body = req.get_json()
        
        # 校验必填字段id
        if "id" not in request_body:
            return func.HttpResponse(
                "请求必须包含id字段",
                status_code=400
            )
        
        # 将JSON转为Document对象，输出绑定自动处理Upsert
        doc.set(func.Document.from_dict(request_body))
        
        return func.HttpResponse("文档已成功更新/创建", status_code=200)
    except Exception as e:
        return func.HttpResponse(f"处理失败: {str(e)}", status_code=500)

关键说明 #

• Upsert逻辑：当传入文档的id（+分区键）与现有文档匹配时，输出绑定自动覆盖该文档；无匹配时创建新文档。
• 数据格式：请求Body需为JSON格式，例如：

  {
    "id": "existing-document-id",
    "partitionKey": "your-partition-value",
    "otherField": "updated-content"
  }
  

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