azure-cosmos-db-py

捆绑的上游支持文件

支持文件位于相对于技能的路径下。有关其固定版本的来源、许可证、哈希值和导入变更，请参阅 UPSTREAM-NOTICE.md。

已恢复辅助工具的安全性

集成测试样例 assets/conftest_template.py 会连接到已配置的 Cosmos 终结点；cleanup_test_docs 会删除明确添加到其跟踪列表中的记录。只能使用隔离且可随时销毁的测试容器，并仅存放测试数据，绝不可使用生产数据。复制此模板不会连接到 Cosmos。

Cosmos DB 服务实现

审查修正（2026-10-23）：这些代码片段是不完整的应用起步模板。请在启动期间、事件循环之外初始化同步客户端，并串行化初始化过程，在关闭时关闭客户端和凭据。对于缓存的初始化失败，需要提供明确且有界的恢复路径。也可使用 azure.cosmos.aio；该 SDK 并非仅支持同步操作。必须独立于分区路由执行租户授权；捆绑客户端假定使用 /workspaceId。绝不可接受调用方提供的 SQL 片段。应区分存储不可用（503）、未找到（404）和空数据；下方较旧的宽容默认值示例演示了一种应避免用于权威读取的策略。使用 ETag 条件写入以防止更新丢失，以原子方式强制执行唯一性，并通过支持冲突处理的补偿机制实现跨分区移动；先读取再执行 upsert/delete 并非原子操作。这些模板未验证任何实际 Cosmos 行为。

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

安装

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 Router                          │
│  - Auth dependencies (get_current_user, get_current_user_required)
│  - HTTP error responses (HTTPException)                         │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                        Service Layer                            │
│  - Business logic and validation                                │
│  - Document ↔ Model conversion                                  │
│  - Graceful degradation when Cosmos unavailable                 │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                     Cosmos DB Client Module                     │
│  - Singleton container initialization                           │
│  - Dual auth: DefaultAzureCredential (Azure) / Key (emulator)   │
│  - Async wrapper via run_in_threadpool                          │
└─────────────────────────────────────────────────────────────────┘

快速开始

1. 客户端模块设置

创建支持双重身份验证的 Cosmos 单例客户端：

# 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:
    from urllib.parse import urlsplit
    parsed = urlsplit(endpoint)
    return (parsed.scheme == "https" and parsed.username is None
            and parsed.password is None
            and parsed.hostname in {"localhost", "127.0.0.1", "::1"})

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 _cosmos_container

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

2. Pydantic 模型层级

使用五层模型模式，实现清晰分离：

class ProjectBase(BaseModel):           # Shared fields
    name: str = Field(..., min_length=1, max_length=200)

class ProjectCreate(ProjectBase):       # Creation request
    workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel):         # Partial updates (all optional)
    name: Optional[str] = Field(None, min_length=1)

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

class ProjectInDB(Project):             # Internal with docType
    doc_type: str = "project"

3. 服务层模式

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)

完整模式：请参阅 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 要求

使用以下模式，在实现之前编写测试：

@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 固件

质量属性（NFR）

可靠性

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

安全性

• 代码中不包含任何密钥（通过 DefaultAzureCredential 使用 RBAC）
• 参数化查询可防止注入
• 分区路由不负责授权租户；接受工作区 ID 前必须验证成员身份和角色

可维护性

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

可测试性

• 通过 get_container() 实现依赖注入
• 使用模块级全局变量即可轻松模拟
• 清晰的分层支持在不使用 Cosmos 的情况下进行单元测试

性能

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

何时使用

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

限制

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