用途
本技能用于构建生产级 Azure Cosmos DB NoSQL 服务,遵循整洁代码、安全最佳实践与 TDD(测试驱动开发)原则,技术栈为 Python,配合 FastAPI 使用。
主要能力
- 提供客户端模块模板:单例容器初始化,支持双重认证(Azure 上用 DefaultAzureCredential,模拟器上用密钥),并通过 run_in_threadpool 做异步封装,避免阻塞 FastAPI 事件循环。
- 提供五层 Pydantic 模型模式:Base(共享字段)、Create(创建请求)、Update(可选字段的部分更新)、响应模型、InDB(含 docType 的内部模型),实现关注点分离与模式演进。
- 提供服务层模式:业务逻辑与校验、文档与模型互转、Cosmos 不可用时优雅降级(返回 None/[]),命名约定如 _doc_to_model()、_model_to_doc()、_use_cosmos()。
- 提供 TDD 测试模式:先写测试,使用 pytest、pytest-asyncio 与 mocker 模拟 Cosmos 容器。
- 附带参考文件:client-setup.md(双重认证、SSL、单例)、service-layer.md(CRUD、转换、降级)、testing.md(pytest、mock、集成测试)、partitioning.md(分区键选择、跨分区查询、移动操作)、error-handling.md(CosmosResourceNotFoundError、日志、HTTP 错误映射)。
- 附带模板文件:cosmos_client_template.py、service_template.py、conftest_template.py。
使用方式
安装依赖:
pip install azure-cosmos azure-identity
环境变量:
COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/ COSMOS_DATABASE_NAME=<database-name> COSMOS_CONTAINER_ID=<container-id>
For emulator only (not production)
COSMOS_KEY=<emulator-key>
认证(优先 DefaultAzureCredential):
from azure.cosmos import CosmosClient from azure.identity import DefaultAzureCredential
client = CosmosClient( url=os.environ["COSMOS_ENDPOINT"], credential=DefaultAzureCredential() )
模拟器(本地开发):
from azure.cosmos import CosmosClient
client = CosmosClient( url="https://localhost:8081", credential=os.environ["COSMOS_KEY"], connection_verify=False )
架构分层为:FastAPI 路由层(认证依赖、HTTP 错误响应)→ 服务层(业务逻辑、文档与模型转换、降级)→ Cosmos DB 客户端模块(单例容器、双重认证、异步封装)。
测试示例:
@pytest.fixture def mock_cosmos_container(mocker): container = mocker.MagicMock() mocker.patch("app.db.cosmos.get_container", return_value=container) return container
核心原则
安全要求:Azure 上使用 DefaultAzureCredential 做 RBAC 认证,绝不在代码中存储密钥;模拟器密钥仅供本地开发硬编码;始终使用 @parameter 参数化查询,禁止字符串拼接;校验分区键访问与用户授权一致。
整洁代码:客户端模块负责连接,服务负责业务逻辑;Cosmos 不可用时降级;统一命名与完整类型标注;使用 Field(alias="camelCase") 做 JSON 序列化。
适用场景
适用于在 FastAPI 中实现基于 Azure Cosmos DB NoSQL 的增删改查服务、服务层封装、分区键设计、错误处理与 pytest 模拟测试,并要求遵循整洁代码与安全实践的 Python 项目。
限制与注意事项
- 集成夹具 assets/conftest_template.py 会连接已配置的 Cosmos 端点,cleanup_test_docs 会删除显式加入其追踪列表的记录;只能使用隔离的一次性测试容器与仅测试用记录,绝不能使用生产数据。复制模板本身不会连接 Cosmos。
- 该技能仅适用于与上述范围明确匹配的任务;输出不能替代针对具体环境的验证、测试或专家审查;若缺少必要输入、权限、安全边界或成功标准,应停止并请求澄清。
- 其他未在资料中提供的功能、收费、平台与效果暂未提供。