SuperAPI AI 工具集

azure-cosmos-db-py

编程与开发

查看项目

更新于 10-16 02:12

技能介绍

用途

本技能用于构建生产级 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。
  • 该技能仅适用于与上述范围明确匹配的任务;输出不能替代针对具体环境的验证、测试或专家审查;若缺少必要输入、权限、安全边界或成功标准,应停止并请求澄清。
  • 其他未在资料中提供的功能、收费、平台与效果暂未提供。