如何在Python编写的Azure Function中调用Cosmos DB存储过程？

首先明确：无需手动构造HTTP请求调用Cosmos DB存储过程，可直接使用官方azure-cosmos Python SDK完成调用，目前Azure Functions没有提供调用存储过程的专用绑定，直接引入SDK操作是最简洁的方案。

步骤1：配置依赖与环境变量 #

• 向你的Function项目requirements.txt添加依赖：

azure-functions
azure-cosmos>=4.4.0

• 在Azure Function的「配置」-「应用程序设置」中添加以下配置项，本地开发则添加到local.settings.json的Values节点：
  • CosmosDbConnectionString：你的Cosmos DB账户连接字符串
  • CosmosDbDatabaseId：目标数据库名称
  • CosmosDbContainerId：目标容器名称
  • CosmosDbSpName：你创建的无参数存储过程名称

步骤2：代码实现 #

import azure.functions as func
from azure.cosmos import CosmosClient
import os

def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        # 初始化Cosmos客户端
        cosmos_conn_str = os.environ["CosmosDbConnectionString"]
        client = CosmosClient.from_connection_string(cosmos_conn_str)
        
        # 获取目标数据库和容器
        database = client.get_database_client(os.environ["CosmosDbDatabaseId"])
        container = database.get_container_client(os.environ["CosmosDbContainerId"])
        
        # 调用无参数存储过程
        # 注意：如果你的容器配置了分区键，必须指定partition_key参数为对应的值
        sp_result = container.scripts.execute_stored_procedure(
            sproc=os.environ["CosmosDbSpName"],
            params=[], # 无参数传空列表即可
            partition_key="你的分区键值" # 如果容器无分区键可删除该参数
        )
        
        return func.HttpResponse(
            f"存储过程执行成功，返回结果：{str(sp_result)}",
            status_code=200
        )
    except Exception as e:
        return func.HttpResponse(
            f"存储过程执行失败：{str(e)}",
            status_code=500
        )

关键注意事项 #

• 如果你的容器启用了分区键，调用execute_stored_procedure时必须传入对应的分区键值，存储过程只会作用于指定分区键对应的数据范围
• 存储过程的执行有超时限制，最长5秒，若有长时任务建议拆分逻辑
• 所用连接字符串需要具备容器的存储过程执行权限，默认读写权限即可满足要求

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