Suporte upstream incluído

Arquivos de suporte são incluídos em caminhos relativos à skill. Veja UPSTREAM-NOTICE.md para suas fontes fixadas, licenças, hashes e alterações de importação.

Segurança de helper restaurada

O fixture de integração em assets/conftest_template.py se conecta ao endpoint Cosmos configurado; cleanup_test_docs exclui registros explicitamente adicionados à sua lista de rastreamento. Use apenas um container de teste isolado e descartável com registros apenas para teste, nunca dados de produção. Copiar o template não se conecta ao Cosmos.

Implementação do Serviço Cosmos DB

Construa serviços Azure Cosmos DB NoSQL de nível de produção seguindo princípios de código limpo, melhores práticas de segurança e TDD.

Instalação

pip install azure-cosmos azure-identity

Variáveis de Ambiente

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>

Autenticação

DefaultAzureCredential (preferencial):

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

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

Emulador (desenvolvimento local):

from azure.cosmos import CosmosClient

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

Visão Geral da Arquitetura

┌─────────────────────────────────────────────────────────────────┐
│                         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                          │
└─────────────────────────────────────────────────────────────────┘

Início Rápido

1. Configuração do Módulo Cliente

Crie um cliente Cosmos singleton com autenticação dupla:

# 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

Implementação completa: Veja references/client-setup.md

2. Hierarquia de Modelos Pydantic

Use o padrão de modelo de cinco níveis para separação limpa:

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

class ProjectCreate(ProjectBase):       # Requisição de criação
    workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel):         # Atualizações parciais (todas opcionais)
    name: Optional[str] = Field(None, min_length=1)

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


…