Azure Function中Cosmos DB单例类导入ModuleNotFound问题及优化建议

一、ModuleNotFoundError 解决方法 #

你的文件夹结构中，function_app.py 位于 functions/fn_role_description/ 目录下，要导入同级目录的 cosmos_client.py，可通过以下两种方式处理：

方案1：使用相对导入 #

修改 function_app.py 中的导入语句：

from ..cosmos_client import CosmosDbClient
from azure.cosmos import exceptions  # 单独导入exceptions，避免路径混淆

方案2：添加项目根目录到Python路径 #

在 function_app.py 开头添加代码，将项目根目录（/webresume/）加入Python搜索路径：

import sys
import os
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))

from functions.cosmos_client import CosmosDbClient
from azure.cosmos import exceptions

注意事项 #

确保所有目录（functions/、fn_role_description/）下的 __init__.py 文件存在（空文件即可），用于标记为Python包。

二、代码优化建议 #

1. CosmosDbClient 单例类优化 #

原单例实现存在逻辑错误和不规范问题，优化后代码如下：

from azure.cosmos import CosmosClient, exceptions
import threading
import os

class CosmosDbClient:
    _instance = None
    _lock = threading.Lock()
    _initialized = False  # 标记是否已完成初始化

    def __new__(cls):
        if not cls._instance:
            with cls._lock:
                if not cls._instance:
                    cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        # 避免重复初始化
        if self._initialized:
            return
        
        try:
            cosmos_url = os.getenv("CosmosDbUri")
            cosmos_key = os.getenv("CosmosDbKey")
            
            if not cosmos_url or not cosmos_key:
                raise ValueError("CosmosDbUri 或 CosmosDbKey 环境变量未设置")
            
            self._client = CosmosClient(cosmos_url, credentials=cosmos_key)
            self._initialized = True
        except Exception as e:
            raise RuntimeError(f"初始化Cosmos客户端失败: {str(e)}") from e

    @property
    def client(self):
        return self._client

优化点说明：

• 修复父类调用错误：原代码错误调用 super(CosmosClient, cls)，改为 super().__new__(cls) 正确调用当前类的父类构造。
• 避免重复初始化：添加 _initialized 标志，防止每次实例化时重复执行 __init__。
• 修正环境变量方法：os.getEnv → os.getenv()（Python标准方法为小写v）。
• 完善错误处理：捕获环境变量缺失的情况，抛出明确错误；移除无效的json.JSONDecodeError捕获（os.getenv不会触发该异常）。
• 简化调用方式：将 cosmos_client() 改为 @property 属性，调用更简洁。

2. Azure Function 代码优化 #

原函数存在逻辑错误、命名不规范等问题，优化后代码如下：

import azure.functions as func
from ..cosmos_client import CosmosDbClient
from azure.cosmos import exceptions
import logging
import json

app = func.FunctionApp()

@app.function_name(name="fn_role_descriptions")
@app.route(route="descriptions", auth_level=func.AuthLevel.FUNCTION)
def role_description_handler(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    # 获取请求中的id参数
    item_id = req.params.get('id')
    if not item_id:
        try:
            req_body = req.get_json()
            item_id = req_body.get('id')
        except ValueError:
            return func.HttpResponse(
                "请求体格式错误，需提供JSON格式数据",
                status_code=400
            )

    if not item_id:
        return func.HttpResponse(
            "请在查询参数或请求体中提供id",
            status_code=400
        )

    try:
        cosmos_service = CosmosDbClient()
        db = cosmos_service.client.get_database_client('webresumedb')
        container = db.get_container_client('descriptions')

        # 使用container执行查询（原代码误用db.query_items）
        items = list(container.query_items(
            query='SELECT * FROM descriptions d WHERE d.Id = @model',
            parameters=[dict(name='@model', value=item_id)],
            enable_cross_partition_query=True  # 容器使用分区键时需开启
        ))

        if items:
            return func.HttpResponse(
                json.dumps(items),
                mimetype="application/json",
                status_code=200
            )
        else:
            return func.HttpResponse(
                json.dumps({"message": "未找到对应记录"}),
                mimetype="application/json",
                status_code=404
            )

    except exceptions.CosmosResourceNotFoundError:
        return func.HttpResponse(
            json.dumps({"message": "数据库或容器不存在"}),
            status_code=404
        )
    except exceptions.CosmosHttpResponseError as e:
        logging.error(f"Cosmos请求错误: {str(e)}")
        return func.HttpResponse(
            json.dumps({"message": "服务器内部错误"}),
            status_code=500
        )
    except Exception as e:
        logging.error(f"未知错误: {str(e)}")
        return func.HttpResponse(
            json.dumps({"message": "服务器内部错误"}),
            status_code=500
        )

优化点说明：

• 移除不必要的类包裹：Azure Functions装饰器直接作用于函数即可，无需封装在类中，简化结构。
• 修正参数获取逻辑：先从查询参数取id，若不存在则从请求体提取，避免原代码中使用空Id查询的错误。
• 修正查询对象：query_items是容器（container）的方法，原代码误用数据库（db）对象调用。
• 完善异常处理：捕获不同类型的Cosmos异常，返回对应HTTP状态码和友好提示，避免直接抛出异常导致500错误。
• 规范变量命名：Id → item_id，符合Python小写加下划线的命名规范。
• 处理查询结果：将查询迭代器转为列表，并根据结果是否存在返回不同响应。
• 添加跨分区查询选项：容器使用分区键时，需开启enable_cross_partition_query=True才能正确查询。

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