---

name: azure-cosmos-db-py description: "构建遵循整洁代码、安全最佳实践和 TDD 原则的生产级 Azure Cosmos DB NoSQL 服务。" risk: unknown source: community date_added: "2026-02-27"

Cosmos DB 服务实现

构建遵循整洁代码、安全最佳实践和 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>
# 仅用于模拟器（非生产环境）
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                          │
│  - 认证依赖 (get_current_user, get_current_user_required)       │
│  - HTTP 错误响应 (HTTPException)                                │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                          服务层                                  │
│  - 业务逻辑和验证                                                │
│  - 文档 ↔ 模型转换                                              │
│  - Cosmos 不可用时的优雅降级                                      │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                   Cosmos DB 客户端模块                            │
│  - 单例容器初始化                                                │
│  - 双重认证：DefaultAzureCredential (Azure) / Key (模拟器)       │
│  - 通过 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:
    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 _cosmos_container

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

2. Pydantic 模型层次结构

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

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"

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. 参数化查询：始终使用参数化查询防止注入攻击