Implémentation du Service Cosmos DB

Construisez des services Azure Cosmos DB NoSQL de qualité production en suivant les principes de code propre, les bonnes pratiques de sécurité et TDD.

Installation

pip install azure-cosmos azure-identity

Variables d'environnement

COSMOS_ENDPOINT=https://<account>.documents.azure.com:443/  # Requis pour toutes les méthodes d'auth
COSMOS_DATABASE_NAME=<database-name>  # Requis pour toutes les méthodes d'auth
COSMOS_CONTAINER_ID=<container-id>  # Requis pour toutes les méthodes d'auth
# Émulateur uniquement (pas en production)
COSMOS_KEY=<emulator-key>  # Requis seulement pour l'auth par clé ou l'émulateur
AZURE_TOKEN_CREDENTIALS=prod # Requis seulement si DefaultAzureCredential est utilisé en production

Authentification

DefaultAzureCredential (préféré) :

import os
from azure.cosmos import CosmosClient
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential

# Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utiliser une credential spécifique directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

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

Émulateur (développement local) :

from azure.cosmos import CosmosClient

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

Vue d'ensemble de l'architecture

┌─────────────────────────────────────────────────────────────────┐
│                       Routeur FastAPI                           │
│  - Dépendances d'auth (get_current_user, get_current_user_required)
│  - Réponses d'erreur HTTP (HTTPException)                       │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                       Couche Service                            │
│  - Logique métier et validation                                 │
│  - Conversion Document ↔ Modèle                                 │
│  - Dégradation gracieuse quand Cosmos est indisponible          │
└──────────────────────────────┬──────────────────────────────────┘
                               │
┌──────────────────────────────▼──────────────────────────────────┐
│                    Module Client Cosmos DB                      │
│  - Initialisation singleton du conteneur                        │
│  - Double auth : DefaultAzureCredential (Azure) / Clé (émulateur)
│  - Wrapper async via run_in_threadpool                          │
└─────────────────────────────────────────────────────────────────┘

Démarrage rapide

1. Configuration du module Client

Créez un client Cosmos singleton avec double authentification :

# 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

Implémentation complète : Voir references/client-setup.md

2. Hiérarchie des modèles Pydantic

Utilisez un motif à cinq niveaux pour une séparation propre :

class ProjectBase(BaseModel):           # Champs partagés
    name: str = Field(..., min_length=1, max_length=200)

class ProjectCreate(ProjectBase):       # Requête de création
    workspace_id: str = Field(..., alias="workspaceId")

class ProjectUpdate(BaseModel):         # Mises à jour partielles (tout optionnel)
    name: Optional[str] = Field(None, min_length=1)

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

class ProjectInDB(Project):             # Interne avec docType
    doc_type: str = "project"

3. Motif de la couche Service

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)

Motifs complets : Voir references/service-layer.md

Principes fondamentaux

Exigences de sécurité

1. Authentification RBAC : Utilisez DefaultAzureCredential dans Azure — ne stockez jamais les clés dans le code
2. Clés émulateur uniquement : Codez en dur la clé d'émulateur bien connue seulement pour le développement local
3. Requêtes paramétrées : Utilisez toujours la syntaxe @parameter — jamais de concaténation de chaînes
4. Validation de clé de partition : Validez que l'accès à la clé de partition correspond à l'autorisation utilisateur

Conventions de code propre

1. Responsabilité unique : Le module client gère la connexion ; les services gèrent la logique métier
2. Dégradation gracieuse : Les services retournent None/[] quand Cosmos est indisponible
3. Nommage cohérent : _doc_to_model(), _model_to_doc(), _use_cosmos()
4. Indices de type : Typage complet sur toutes les méthodes publiques
5. Alias CamelCase : Utilisez Field(alias="camelCase") pour la sérialisation JSON

Exigences TDD

Écrivez les tests AVANT l'implémentation en utilisant ces motifs :

@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"

Guide de test complet : Voir references/testing.md

Fichiers de référence

Fichiers template

Attributs de qualité (NFR)

Fiabilité

• Dégradation gracieuse quand Cosmos est indisponible
• Logique de retry avec backoff exponentiel pour les défaillances transitoires
• Connection pooling via motif singleton

Sécurité

• Zéro secrets dans le code (RBAC via DefaultAzureCredential)
• Requêtes paramétrées préviennent l'injection
• Isolation de clé de partition renforce les limites de données

Maintenabilité

• Motif de modèle à cinq niveaux permet l'évolution du schéma
• Couche service découple la logique métier du stockage
• Motifs cohérents dans tous les services d'entité

Testabilité

• Injection de dépendances via get_container()
• Moquage facile avec des globales au niveau du module
• Séparation claire permet les tests unitaires sans Cosmos

Performance

• Requêtes de clé de partition évitent les scans cross-partition
• Wrapping async empêche le blocage de la boucle d'événements FastAPI
• Surcharge minimale de conversion de document