Azure Cosmos DB Python SDK

azure-cosmos-db-py
分类编程
作者Agentic Awesome Skills 社区
许可MIT
评分4.50/5
使用5.1K

Cosmos DB 服务实现

遵循整洁代码、安全最佳实践和 TDD 原则，构建生产级 Azure Cosmos DB NoSQL 服务。

安装

bash
pip install azure-cosmos azure-identity

环境变量

bash
COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/
COSMOS_DATABASE_NAME=<database-name>
COSMOS_CONTAINER_ID=<container-id>

仅用于模拟器（非生产环境）

COSMOS_KEY=<emulator-key>

身份验证

DefaultAzureCredential (推荐):

python
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential


client = CosmosClient(
    url=os.environ["COSMOS_ENDPOINT"],
    credential=DefaultAzureCredential()
)

模拟器 (本地开发):

python
from azure.cosmos import CosmosClient


client = CosmosClient(
    url="https://localhost:8081",
    credential=os.environ["COSMOS_KEY"],
    connection_verify=False
)

架构概览

code
┌─────────────────────────────────────────────────────────────────┐
│                         FastAPI Router                          │
│  - 认证依赖 (get_current_user, get_current_user_required)        │
│  - HTTP 错误响应 (HTTPException)                                 │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                        Service Layer                            │
│  - 业务逻辑与验证                                               │
│  - 文档 ↔ 模型 转换                                             │
│  - Cosmos 不可用时的优雅降级                                     │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                     Cosmos DB Client Module                     │
│  - 单例容器初始化                                               │
│  - 双重认证：DefaultAzureCredential (Azure) / Key (模拟器)       │
│  - 通过 run_in_threadpool 实现异步封装                           │
└─────────────────────────────────────────────────────────────────┘

快速上手

1. 客户端模块设置

创建具有双重认证功能的 Cosmos 客户端单例：

python
# db/cosmos.py
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential
from starlette.concurrency import run_in_threadpool

_cosmos_container = None


def _is_emulator_endpoint(endpoint: str) -> bool:
    return "localhost" in endpoint or "127.0.0.1" in endpoint


async def get_container():
    global _cosmos_container
    if _cosmos_container is None:
        if _is_emulator_endpoint(settings.cosmos_endpoint):
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=settings.cosmos_key,
                connection_verify=False
            )
        else:
            client = CosmosClient(
                url=settings.cosmos_endpoint,
                credential=DefaultAzureCredential()
            )
        db = client.get_database_client(settings.cosmos_database_name)
        _cosmos_container = db.get_container_client(settings.cosmos_container_id)
    return _co

smos_container
code
完整实现：请参阅 references/client-setup.md


2. Pydantic 模型层级


采用五层模型模式以实现清晰的分离：
python
class ProjectBase(BaseModel): # 共享字段
name: str = Field(..., min_length=1, max_length=200)

class ProjectCreate(ProjectBase): # 创建请求
workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel): # 部分更新（全部可选）
name: Optional[str] = Field(None, min_length=1)

class Project(ProjectBase): # API 响应
id: str
created_at: datetime = Field(..., alias="createdAt")

class ProjectInDB(Project): # 包含 docType 的内部模型
doc_type: str = "project"

code
### 3. 服务层模式 (Service Layer Pattern)
python
class ProjectService:
def _use_cosmos(self) -> bool:
return get_container() is not None

async def get_by_id(self, project_id: str, workspace_id: str) -> Project | None:
if not self._use_cosmos():
return None
doc = await get_document(project_id, partition_key=workspace_id)
if doc is None:
return None
return self._doc_to_model(doc)
code
完整模式：请参阅 references/service-layer.md


核心原则


安全要求


1. RBAC 认证：在 Azure 中使用 DefaultAzureCredential —— 绝不要在代码中存储密钥。
2. 模拟器专用密钥：仅在本地开发时硬编码已知的模拟器密钥。
3. 参数化查询：始终使用 @parameter 语法 —— 绝不要使用字符串拼接。
4. 分区键验证：验证分区键的访问权限与用户授权相匹配。


代码规范


1. 单一职责：客户端模块处理连接；服务层处理业务逻辑。
2. 优雅降级：当 Cosmos 不可用时，服务应返回 None 或 []。
3. 统一命名：使用 _doc_to_model(), _model_to_doc(), _use_cosmos()。
4. 类型提示：所有公共方法必须包含完整的类型标注。
5. 驼峰式别名：使用 Field(alias="camelCase") 进行 JSON 序列化。


TDD 要求


在实现之前，请使用以下模式编写测试：
python
@pytest.fixture
def mock_cosmos_container(mocker):
container = mocker.MagicMock()
mocker.patch("app.db.cosmos.get_container", return_value=container)
return container

@pytest.mark.asyncio
async def test_get_project_by_id_returns_project(mock_cosmos_container):
# Arrange (准备)
mock_cosmos_container.read_item.return_value = {"id": "123", "name": "Test"}

# Act (执行)
result = await project_service.get_by_id("123", "workspace-1")

# Assert (断言)
assert result.id == "123"
assert result.name == "Test"
``

完整测试指南：请参阅 references/testing.md

参考文件

| 文件 | 阅读时机 |
|------|--------------|
| references/client-setup.md | 配置具有双重认证、SSL 配置和单例模式的 Cosmos 客户端时 |
| references/service-layer.md | 实现包含 CRUD、转换和优雅降级的完整服务类时 |
| references/testing.md | 编写 pytest 测试、模拟 Cosmos 或设置集成测试时 |
| references/partitioning.md | 选择分区键、执行跨分区查询或迁移操作时 |
| references/error-handling.md | 处理 CosmosResourceNotFoundError、日志记录或 HTTP 错误映射时 |

模板文件

| 文件 | 用途 |
|------|---------|
| assets/cosmos_client_template.py | 即插即用的客户端模块 |
| assets/service_template.py | 服务类模板 |
类骨架 |
| assets/conftest_template.py | 用于 Cosmos 模拟的 pytest fixtures |

质量属性 (NFRs)

可靠性

  • Cosmos 不可用时的优雅降级
  • 针对瞬时故障的指数退避重试逻辑
  • 通过单例模式实现连接池

安全性

  • 代码中零密钥 (通过 DefaultAzureCredential 实现 RBAC)
  • 参数化查询防止注入
  • 分区键隔离强制执行数据边界

可维护性

  • 五层模型模式支持模式 (Schema) 演进
  • 服务层将业务逻辑与存储解耦
  • 所有实体服务采用一致的模式

可测试性

  • 通过 get_container()` 实现依赖注入
  • 通过模块级全局变量轻松进行模拟 (Mocking)
  • 清晰的分离使得在无需 Cosmos 的情况下进行单元测试

性能

  • 分区键查询避免跨分区扫描
  • 异步封装防止阻塞 FastAPI 事件循环
  • 极低的文档转换开销

适用场景

当需要执行概述中描述的工作流或操作时，可使用此技能。

局限性

  • 仅在任务明确符合上述范围时使用此技能。
  • 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。
  • 如果缺少必要的输入、权限、安全边界或成功标准，请停止并寻求澄清。