Azure Functions Python v2版Http Trigger中复用Cosmos DB客户端的正确语法咨询

我来帮你分析这个问题～你遇到的函数检测失败的问题，大概率是全局代码执行时抛出了未被捕获的异常，导致整个Azure Function应用启动失败，自然就检测不到你的customers函数了。

先看你原来的全局代码里的一个明显问题：你导入了DefaultAzureCredential，但并没有创建它的实例，直接用了credential变量，这会触发NameError，肯定会导致启动失败。另外，全局代码在Function App启动阶段就会执行，如果环境变量没配置、Cosmos账号连接失败或者认证出问题，都会直接导致应用启动崩溃，函数也就无法被加载。

按照最佳实践，我们确实应该复用全局的Cosmos客户端，但要确保全局初始化的代码是健壮的，或者采用延迟初始化的方式（第一次调用函数时再初始化客户端），这样既能复用连接，又能避免启动阶段的异常导致函数无法被检测到。

接下来给你两种正确的实现方式：

方式一：修复全局初始化错误+添加异常处理 #

先把你代码里的Credential实例化问题解决，再给全局初始化代码加上异常捕获，确保即使初始化失败，Function App也能正常启动，后续可以在函数里处理错误：

import azure.functions as func
import logging
from azure.cosmos import CosmosClient, exceptions
from azure.identity import DefaultAzureCredential
import os

# 先声明全局变量，初始化为None
cosmos_client = None
cosmos_database = None
cosmos_container = None

try:
    # 初始化Cosmos相关资源
    cosmos_account = os.environ.get("COSMOS_ACCOUNT")
    if not cosmos_account:
        raise ValueError("COSMOS_ACCOUNT环境变量未配置")
    
    credential = DefaultAzureCredential()  # 这里要创建Credential实例！
    cosmos_client = CosmosClient(f"https://{cosmos_account}.documents.azure.com:443/", credential)
    
    cosmos_db_name = os.environ.get("COSMOS_DATABASE")
    if not cosmos_db_name:
        raise ValueError("COSMOS_DATABASE环境变量未配置")
    cosmos_database = cosmos_client.get_database_client(cosmos_db_name)
    
    cosmos_container_name = os.environ.get("COSMOS_CONTAINER")
    if not cosmos_container_name:
        raise ValueError("COSMOS_CONTAINER环境变量未配置")
    cosmos_container = cosmos_database.get_container_client(cosmos_container_name)
    
    logging.info("Cosmos DB客户端初始化成功")
except Exception as e:
    logging.error(f"Cosmos DB客户端初始化失败: {str(e)}")
    # 捕获异常不抛出，保证Function App能正常启动
    pass

app = func.FunctionApp()

@app.route(route="customers/{id:guid?}", auth_level=func.AuthLevel.FUNCTION, methods=["GET", "POST", "PUT", "DELETE"])
def customers(req: func.HttpRequest) -> func.HttpResponse:
    # 在函数里先检查客户端是否初始化成功
    if not cosmos_container:
        return func.HttpResponse("Cosmos DB客户端未初始化", status_code=500)
    
    # 后续正常使用cosmos_container操作数据
    try:
        customer_id = req.route_params.get("id")
        if customer_id:
            item = cosmos_container.read_item(item=customer_id, partition_key=customer_id)
            return func.HttpResponse(f"获取到客户: {item}", status_code=200)
        # 可扩展处理POST/PUT/DELETE等其他请求逻辑
        # ...
    except exceptions.CosmosResourceNotFoundError:
        return func.HttpResponse("客户不存在", status_code=404)
    except Exception as e:
        logging.error(f"处理请求失败: {str(e)}")
        return func.HttpResponse("服务器内部错误", status_code=500)

方式二：延迟初始化（推荐） #

这种方式把客户端初始化逻辑移到函数内部，第一次调用时才执行，既保证了客户端的复用（全局变量只初始化一次），又避免了启动阶段的异常影响函数加载，还通过线程锁保证了多并发场景下的线程安全：

import azure.functions as func
import logging
from azure.cosmos import CosmosClient, exceptions
from azure.identity import DefaultAzureCredential
import os
import threading

# 全局变量+锁，确保线程安全的初始化
cosmos_container = None
_init_lock = threading.Lock()

app = func.FunctionApp()

@app.route(route="customers/{id:guid?}", auth_level=func.AuthLevel.FUNCTION, methods=["GET", "POST", "PUT", "DELETE"])
def customers(req: func.HttpRequest) -> func.HttpResponse:
    global cosmos_container
    
    # 双重检查锁，确保只初始化一次
    if not cosmos_container:
        with _init_lock:
            if not cosmos_container:
                try:
                    cosmos_account = os.environ["COSMOS_ACCOUNT"]
                    credential = DefaultAzureCredential()
                    cosmos_client = CosmosClient(f"https://{cosmos_account}.documents.azure.com:443/", credential)
                    cosmos_database = cosmos_client.get_database_client(os.environ["COSMOS_DATABASE"])
                    cosmos_container = cosmos_database.get_container_client(os.environ["COSMOS_CONTAINER"])
                    logging.info("Cosmos DB客户端延迟初始化成功")
                except Exception as e:
                    logging.error(f"Cosmos DB客户端初始化失败: {str(e)}")
                    return func.HttpResponse("服务初始化失败", status_code=503)
    
    # 后续正常使用cosmos_container处理业务
    try:
        customer_id = req.route_params.get("id")
        if customer_id:
            item = cosmos_container.read_item(item=customer_id, partition_key=customer_id)
            return func.HttpResponse(f"客户信息: {item}", status_code=200)
        # 处理其他请求方法的逻辑
        # ...
    except exceptions.CosmosResourceNotFoundError:
        return func.HttpResponse("未找到该客户", status_code=404)
    except Exception as e:
        logging.error(f"请求处理出错: {str(e)}")
        return func.HttpResponse("服务器内部错误", status_code=500)

额外注意事项 #

• 确保你的Function App（本地调试时是local.settings.json，部署到Azure后是应用配置）已经配置了正确的COSMOS_ACCOUNT、COSMOS_DATABASE、COSMOS_CONTAINER环境变量
• 部署到Azure时，推荐给Function App配置Managed Identity权限，让DefaultAzureCredential可以自动完成Cosmos DB的认证
• 两种方式都遵循了Cosmos DB客户端复用的最佳实践，避免了每次调用函数都创建新客户端带来的性能损耗

备注：内容来源于stack exchange，提问作者Daryl Clark