Python
Django
Fastapi
Senior
Celery
Django-Signals
Django-Cache
Websockets-Fastapi
Microservices-Python
Entretien-Python-Senior
Qcm-Django-Avance
Circuit-Breaker
Cqrs-Django
Django-Middleware
📝
Python Django/FastAPI Senior — Architecture & Perf
20 questions niveau senior : cache Django, Celery, signals, middlewares custom, WebSockets FastAPI, microservices, circuit breaker et observabilité. Pour experts Python backend.
Banque de révision : les 60 questions corrigées
Voici l'intégralité des 60 questions de « Python Django/FastAPI Senior — Architecture & Perf », chacune avec sa bonne réponse et son explication. Le quiz interactif ci-dessus en tire une sélection aléatoire à chaque tentative — utilisez cette liste complète pour réviser.
Explication :
Django propose plusieurs niveaux de cache : (1) per-site :
UpdateCacheMiddleware + FetchFromCacheMiddleware. (2) per-view : @cache_page(60 * 15). (3) per-fragment : {% cache 900 'ma_section' %}. Backend Redis : django-redis recommandé en prod. Invalidation : cache.delete('key'), cache.clear().
Mentionner les patterns d'invalidation : time-based (TTL), event-based (signaux post_save pour invalider le cache d'un objet modifié). Redis pour la persistance et les structures avancées.
Explication :
Un middleware Django est une classe avec des méthodes
__init__(get_response) et __call__(request). Exemple : class TimingMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
start = time.time()
response = self.get_response(request)
response['X-Time'] = time.time() - start
return responseAjout dans MIDDLEWARE (l'ordre est important : top-down pour request, bottom-up pour response).
L'ordre des middlewares est critique : SecurityMiddleware doit être en premier, SessionMiddleware avant AuthenticationMiddleware, etc.
Explication :
Pattern JWT FastAPI : (1)
POST /token → vérifier credentials → créer access_token avec jose.jwt.encode() + secret + expiry. (2) Dépendance get_current_user(token: str = Depends(oauth2_scheme)) → décoder le token → retourner l'utilisateur. (3) Routes protégées : Depends(get_current_user). Library : python-jose[cryptography] + passlib[bcrypt].
En prod : utiliser RS256 (asymétrique) au lieu de HS256 (symétrique) pour les microservices — le service resource peut valider sans connaître le secret. Refresh tokens : stocker en BDD avec révocation possible.
Explication :
Optimisations clés : (1) ORM :
select_related(), prefetch_related(), only()/defer(), values(). (2) Cache : Redis pour les QuerySets fréquents. (3) DB : index DB, connection pooling (pgbouncer). (4) Serveur : Gunicorn + workers = 2×CPU+1. (5) Async : tâches longues → Celery. (6) Profiling : django-silk, django-debug-toolbar.
Mentionner l'importance du profiling avant d'optimiser : mesurer d'abord avec silk/debug-toolbar pour identifier les vraies goulots d'étranglement.
Explication :
1. La requête arrive au WSGI/ASGI handler. 2. Passe par les middlewares en ordre (chaque
process_request). 3. URL resolver : match la première URL dans urlpatterns. 4. View : traite la logique métier. 5. Retour via les middlewares en ordre inverse (process_response). 6. Réponse HTTP envoyée au client.
Si un middleware retourne une réponse prématurément (auth failure, rate limit), les middlewares suivants et la view ne sont pas exécutés.
Explication :
Architecture Celery + Django : (1) Broker : Redis ou RabbitMQ comme file de messages. (2) Task :
@shared_task
def send_email(user_id): ... dans tasks.py. (3) Appel : send_email.delay(user.id) → non-bloquant. (4) Worker : celery -A projet worker -l info. (5) Monitoring : Flower. (6) Scheduler : celery beat pour les tâches périodiques.
Bonnes pratiques : idempotence des tâches (en cas de retry), gestion des erreurs avec max_retries, chord/group pour les tâches parallèles.
Explication :
Le Repository Pattern isole les accès BDD dans des classes dédiées :class ArticleRepository:
def __init__(self, db: Session): self.db = db
def get_all(self): return self.db.query(Article).all()
def create(self, data: ArticleCreate): ...La route injecte le repo via
Depends(). Avantages : les routes ne connaissent pas SQLAlchemy, facile à mocker en tests, changement d'ORM transparent.
Combiner avec Unit of Work pattern pour gérer les transactions complexes impliquant plusieurs repositories.
Explication :
Sécurité FastAPI : (1) Injection : SQLAlchemy paramétré (pas de SQL brut). (2) Auth : JWT + OAuth2, pas de tokens longue durée. (3) Rate limiting :
slowapi. (4) CORS : allow_origins strict. (5) Headers : SecurityHeaders middleware. (6) Secrets : pydantic-settings depuis .env. (7) Validation : Pydantic strict mode. (8) Swagger : désactiver en prod si non nécessaire (docs_url=None).
Mentionner HTTPS (géré par le reverse proxy Nginx/Caddy), pas par FastAPI directement.
Explication :
CBV héritent de
View et organisent la logique : def get(), def post(). Generic CBV intégrés : ListView, DetailView, CreateView, UpdateView, DeleteView. Mixins : LoginRequiredMixin, PermissionRequiredMixin. Avantages : DRY (moins de code répété), extensibilité, cohérence.
Inconvénient : la chaîne MRO (Method Resolution Order) peut être difficile à déboguer. Recommandation : préférer les FBV pour les vues simples, CBV pour les CRUD standardisés.
Explication :
from fastapi import WebSocket
@app.websocket('/ws')
async def ws_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f'Echo: {data}')FastAPI supporte les WebSockets nativement via Starlette. Gestion des déconnexions avec
WebSocketDisconnect.
Explication :
Bonnes pratiques : (1) Nommage :
makemigrations --name 'add_article_slug'. (2) Conflicts : Django détecte les migrations conflictuelles ; makemigrations --merge les résout. (3) Squash : squashmigrations app 0001 0020 pour simplifier l'historique. (4) Atomic : les migrations sont atomiques par défaut. (5) CI : vérifier que les migrations sont à jour (migrate --check).
Ne jamais modifier une migration déjà appliquée en production. Créer une nouvelle migration corrective.
Explication :
from fastapi import BackgroundTasks
@app.post('/send-email/')
def send_email(bg: BackgroundTasks, email: str):
bg.add_task(send_notification, email)
return {'message': 'Email en cours d\'envoi'}La réponse 200 est envoyée immédiatement,
send_notification s'exécute après. Idéal pour des tâches légères. Pour des tâches lourdes ou retry : Celery.
Limitation : les BackgroundTasks FastAPI n'ont pas de persistance (crash = tâche perdue). Pour les tâches critiques, utiliser Celery + Redis.
Explication :
L'architecture hexagonale isole le domaine pur (entités, use cases) des détails techniques. Ports : interfaces Python abstraites (ABC). Adapters : implémentations concrètes (SQLAlchemy, DRF serializer, Pydantic). Le domaine ne dépend pas de Django/FastAPI → testable sans BDD, changeable d'ORM. Couches : Domain → Application (use cases) → Infrastructure (adapters).
Utile pour les gros projets. Pour les petits projets : over-engineering. Mentionner DDD (Domain-Driven Design) comme approche complémentaire.
Explication :
LOGGING = {
'version': 1,
'handlers': {
'file': {'class': 'logging.FileHandler', 'filename': 'django.log'},
'sentry': {'class': 'raven.contrib.django.raven_compat.SentryHandler'}
},
'loggers': {'django': {'handlers': ['file', 'sentry'], 'level': 'ERROR'}}
}En production : Sentry pour les erreurs, ELK/Datadog pour les logs structurés (JSON).
Explication :
Architecture prod : Nginx (port 80/443, TLS, rate limiting, static files) → Uvicorn/Gunicorn (workers = 2×CPU+1,
uvicorn.workers.UvicornWorker) → FastAPI app. Dockerfile : FROM python:3.12-slim; COPY requirements.txt; RUN pip install -r requirements.txt; CMD ["uvicorn", "main:app", "--host", "0.0.0.0"].
Mentionner les variables d'environnement injectées via docker-compose ou Kubernetes secrets, jamais dans l'image. Healthcheck endpoint pour Docker/K8s.
Explication :
CQRS : séparer les modèles de lecture et d'écriture. En Django : (1) Commands : actions métier (CreateArticleCommand) → handlers qui écrivent en BDD. (2) Queries : lecture optimisée (GetArticleListQuery) → peut utiliser
values(), sous-requêtes, vues SQL, cache. Avantage : chaque côté optimisé indépendamment. Surtout utile avec Event Sourcing.
Pour la plupart des projets Django, CQRS est over-engineering. Commencer par l'architecture standard et migrer si les besoins le justifient.
Explication :
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.get('/items/')
@limiter.limit('100/minute')
async def list_items(request: Request): ...Requiert Redis comme backend. Pour la prod : rate limiting au niveau Nginx (plus performant).
Explication :
(1) SQL Injection : L'ORM Django utilise des requêtes paramétrées (jamais de concaténation SQL). (2) XSS : Le système de templates auto-échappe le HTML (
{{ var }} → entités HTML). Pour du HTML intentionnel : {{ var|safe }} (utiliser avec précaution). (3) CSRF : CsrfViewMiddleware + {% csrf_token %}. (4) Clickjacking : X-Frame-Options: DENY (XFrameOptionsMiddleware).
Mentionner aussi la commande
python manage.py check --deploy qui liste les problèmes de sécurité avant le déploiement.
Explication :
Architecture microservices FastAPI : (1) Services indépendants avec leur BDD (database per service). (2) Communication : HTTP sync (httpx) ou async (Kafka/RabbitMQ). (3) API Gateway (Nginx/Kong) devant les services. (4) Service discovery (Consul/K8s). (5) Chaque service : Dockerfile + CI/CD indépendants. Avantage FastAPI : léger, async natif, démarrage rapide.
Mentionner les patterns : Circuit Breaker, Saga pour les transactions distribuées, Event Sourcing pour la consistance éventuelle.
Explication :
Deux approches : (1)
CONN_MAX_AGE = 60 dans DATABASES : Django réutilise la connexion pendant 60 secondes par thread (persistent connections). (2) pgBouncer : proxy de pooling externe (recommandé en prod pour Gunicorn multi-process). Évite le thundering herd au démarrage. Avec CONN_MAX_AGE=0 : nouvelle connexion à chaque requête.
Explication :
Pyramide de tests : (1) Unit : fonctions pures, serializers, validators — pytest + unittest.mock. (2) Integration : routes + BDD réelle — TestClient (FastAPI) ou test.Client (Django) +
pytest-django + BDD test. (3) E2E : scénarios complets. Tools : factory_boy (fixtures), faker (données), pytest-cov (coverage), freezegun (dates fixes).
Bonnes pratiques : tests indépendants, pas de dépendances entre tests, bases de test isolées (Django crée une DB test temporaire), coverage objectif 80%+ sur la logique métier.
Explication :
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
debug: bool = False
class Config:
env_file = '.env'
settings = Settings()Injection via
Depends(get_settings) pour l'override en tests. Type-safe, validé au démarrage.
Explication :
from sqlalchemy.orm import selectinload
result = await db.execute(select(Article).options(selectinload(Article.tags)))
articles = result.scalars().all(). joinedload = JOIN SQL (bon pour une relation), selectinload = IN query (bon pour ManyToMany). En async SQLAlchemy : utiliser lazy='subquery' ou les options explicites.
Explication :
Stack d'observabilité : (1) Logging :
loguru ou structlog pour les logs JSON structurés. (2) Metrics : prometheus-fastapi-instrumentator → expose /metrics pour Prometheus/Grafana. (3) Tracing : opentelemetry-sdk + opentelemetry-instrumentation-fastapi → Jaeger/Zipkin. (4) Error tracking : Sentry (sentry-sdk[fastapi]).
Mentionner la corrélation des traces avec des correlation IDs propagés dans les headers X-Correlation-ID pour suivre une requête à travers les microservices.
Explication :
Optimisations admin gros volumes : (1)
show_full_result_count = False : évite le COUNT(*) coûteux. (2) list_select_related = True : réduit les N+1. (3) raw_id_fields = ('auteur',) : remplace les dropdown (qui font un SELECT ALL) par un champ ID. (4) search_fields avec des index BDD appropriés. (5) list_per_page = 25.
Explication :
WSGI (PEP 3333) : interface standard synchrone. Un thread/process = une requête. Serveurs : Gunicorn, uWSGI. ASGI (PEP 595) : interface async. Un process peut gérer de nombreuses connexions concurrentes (WebSockets, SSE, HTTP/2). Serveurs : Uvicorn, Daphne, Hypercorn. Django supporte les deux (WSGI depuis toujours, ASGI depuis 3.0). FastAPI = ASGI natif.
Pour Django avec des views async, utiliser ASGI (Daphne ou Uvicorn) — WSGI ne peut pas exécuter du code async nativement.
Explication :
La pagination par curseur est plus efficace que offset pour les grandes tables (évite un scan séquentiel). Exemple :
WHERE id > {cursor} ORDER BY id LIMIT {limit}. La réponse inclut le next_cursor (dernier id renvoyé). Avantages : stable (pas de saut si nouvelles lignes), performant (index scan). fastapi-pagination supporte les curseurs.
Mentionner quand utiliser cursor vs offset : cursor pour les feeds temps réel, offset pour les résultats de recherche où l'utilisateur peut sauter à une page arbitraire.
Explication :
Pattern moderne (FastAPI 0.93+) :from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup : init DB, warm cache, connect broker
db_pool = await create_pool(DATABASE_URL)
yield # L'app tourne ici
# Shutdown : fermer connexions
await db_pool.close()
app = FastAPI(lifespan=lifespan)
Explication :
def override_get_current_user():
return User(id=1, email='test@test.com')
app.dependency_overrides[get_current_user] = override_get_current_user
client = TestClient(app)
# Tests...
app.dependency_overrides = {} # NettoyagePermettant de tester les routes protégées sans vraie auth.
Explication :
Stratégie robuste : (1)
@app.exception_handler(HTTPException) : formater les erreurs HTTP. (2) @app.exception_handler(RequestValidationError) : personnaliser les 422. (3) @app.exception_handler(Exception) : catch-all → log + 500. (4) Exceptions métier personnalisées héritant de HTTPException. (5) Ne jamais exposer les stack traces en production. (6) Sentry pour la capture des erreurs 5xx.
Adopter une structure de réponse d'erreur cohérente : {\"error\": \"code\", \"message\": \"...\", \"details\": [...]} pour faciliter le débogage côté client.
Explication :
class ArticleManager(models.Manager):
def publies(self):
return self.filter(statut='publié')
def par_auteur(self, auteur):
return self.filter(auteur=auteur)
class Article(models.Model):
objects = ArticleManager()Usage :
Article.objects.publies().par_auteur(user). Encapsule la logique de requête, améliore la lisibilité et la réutilisabilité.
Explication :
class IsOwnerOrReadOnly(permissions.BasePermission):
def has_object_permission(self, request, view, obj):
if request.method in SAFE_METHODS:
return True
return obj.auteur == request.user
has_permission = niveau view (liste). has_object_permission = niveau objet individuel. Combiner : permission_classes = [IsAuthenticated, IsOwnerOrReadOnly].
Explication :
class PaginationParams:
def __init__(self, skip: int = 0, limit: int = 10):
self.skip = skip
self.limit = limit
@app.get('/items/')
def list_items(pagination: PaginationParams = Depends()):
# pagination.skip, pagination.limit
...FastAPI instancie la classe et injecte les paramètres query automatiquement.
Explication :
Stratégie zero-downtime pour ajouter une colonne
NOT NULL : (1) Ajouter la colonne avec null=True, blank=True + migrer (compatible avec l'ancienne version de l'app). (2) Déployer la nouvelle version de l'app qui remplit le champ. (3) Backfill : Article.objects.filter(slug='').update(slug=...). (4) Ajouter la contrainte NOT NULL + migration. (5) Déployer la migration finale.
Bibliothèque
django-safedelete et django-zero-downtime-migrations aident à identifier les migrations potentiellement bloquantes (ajout d'index sur grande table).
Explication :
from django.db.models import OuterRef, Subquery
latest_comment = Comment.objects.filter(
article=OuterRef('pk')
).order_by('-created').values('contenu')[:1]
articles = Article.objects.annotate(
last_comment=Subquery(latest_comment)
)Génère une sous-requête SQL corrélée. Plus expressif que
raw() et portable entre BDD.
Explication :
from sse_starlette.sse import EventSourceResponse
@app.get('/events')
async def sse_endpoint(request: Request):
async def event_generator():
while True:
if await request.is_disconnected(): break
yield {'data': json.dumps({'time': time.time()})}
await asyncio.sleep(1)
return EventSourceResponse(event_generator())SSE : unidirectionnel server → client, plus simple que WebSockets pour les flux de données.
Explication :
bulk_create() insère en une seule requête SQL INSERT (ou par batchs). Paramètre batch_size=500 pour les très grandes listes. Limitations : ne déclenche pas les signaux post_save, ne gère pas les contraintes d'unicité par défaut (sauf update_conflicts=True en Django 4.1+).
Explication :
Outils : (1) Debug :
django-debug-toolbar → panel SQL (nb requêtes, temps, stacktrace). (2) Code : from django.db import connection; print(connection.queries). (3) Staging : django-silk → profiling complet des requêtes. (4) SQL : Article.objects.filter(...).explain() → EXPLAIN en Django 2.1+. (5) Prod : pgBadger, slow query log PostgreSQL.
Explication :
def get_db():
db = SessionLocal()
try:
yield db # Fourni à la route
finally:
db.close() # Exécuté après la réponseLe
finally garantit la fermeture même si une exception est levée. FastAPI gère automatiquement l'appel de close() après chaque requête.
Explication :
Stratégies : (1) URL versioning (recommandé) :
/api/v1/items, /api/v2/items — routers APIRouter séparés. (2) Header versioning : Accept: application/vnd.api+json; version=1. (3) Dépréciation : marquer les routes dépréciées dans la doc (deprecated=True dans le décorateur). (4) Maintenir au moins N-1 versions actives.
Mentionner la documentation OpenAPI multi-version : générer un /openapi-v1.json et /openapi-v2.json séparés pour des clients distincts.
Explication :
F('vues') + 1 génère un UPDATE SQL atomique : UPDATE articles SET vues = vues + 1 WHERE id = 1. Sans F(), le pattern get() → vues += 1 → save() crée une race condition (deux requêtes concurrentes écrasent la valeur l'une de l'autre). F() résout ce problème au niveau BDD.
Explication :
Architecture HA Django : (1) Load Balancer : Nginx/HAProxy/AWS ALB avec health checks. (2) App servers : N instances Gunicorn (stateless, docker/k8s). (3) BDD : PostgreSQL primaire + read replicas, pgBouncer pour pooling. (4) Cache : Redis cluster. (5) Async : Celery workers + beat. (6) Static : S3/CDN. (7) Sessions : Redis (pas en BDD). (8) Monitoring : Prometheus/Grafana + Sentry.
Clé : rendre les workers Django 100% stateless (pas de fichiers locaux, sessions en Redis, aucune config en mémoire partagée).
Explication :
Bibliothèques recommandées :
authlib (OAuth1/2/OIDC) ou fastapi-users (auth complète avec OAuth2). Flow : (1) Rediriger vers provider/authorize. (2) Recevoir le code de retour (/callback). (3) Échanger le code contre un access_token. (4) Récupérer le profil utilisateur. (5) Créer/mettre à jour l'utilisateur en BDD. (6) Émettre un JWT interne.
Explication :
Pattern soft delete : (1) Ajouter
is_deleted = models.BooleanField(default=False) + deleted_at. (2) Créer un SoftDeleteManager qui filtre is_deleted=False par défaut. (3) Surcharger delete() : self.is_deleted = True; self.save(). (4) Ajouter un hard_delete() pour la vraie suppression. Bibliothèque : django-safedelete.
Mentionner les implications : contraintes d'unicité (un email unique peut bloquer la recréation) → partial unique index en PostgreSQL pour exclure les soft-deleted.
Explication :
from pydantic import model_validator
class EventCreate(BaseModel):
start: datetime
end: datetime
@model_validator(mode='after')
def check_dates(self) -> 'EventCreate':
if self.end
Explication :
Event Sourcing : au lieu de stocker
account.balance = 1000, on stocke les événements [DepositEvent(200), WithdrawEvent(100)]. L'état courant est reconstruit en rejouant les événements. Avantages : historique complet, audit log, time-travel debugging. Bibliothèques Python : eventsourcing. Complexité accrue — justifié pour les domaines avec audit strict.
Event Sourcing + CQRS souvent combinés. Mentionner les défis : migration des événements, performance (replay long), idempotence des event handlers.
Explication :
class ArticleCreateView(
LoginRequiredMixin,
PermissionRequiredMixin,
CreateView
):
model = Article
form_class = ArticleForm
permission_required = 'blog.add_article'
success_url = reverse_lazy('article-list')
def form_valid(self, form):
form.instance.auteur = self.request.user
return super().form_valid(form)
Explication :
@shared_task(bind=True, max_retries=3, autoretry_for=(Exception,), retry_backoff=True)
def send_email(self, user_id):
try:
...
except SMTPError as exc:
raise self.retry(exc=exc, countdown=2**self.request.retries)
retry_backoff=True : délai exponentiel (2^retry) évite le thundering herd. max_retries=3 : max 3 tentatives.
Explication :
En production, Nginx (ou Caddy, Traefik) gère le TLS, puis proxifie vers Uvicorn en HTTP interne. Avantages : meilleure performance TLS, gestion des certificats Let's Encrypt (Certbot), termination TLS + load balancing. En dev :
uvicorn main:app --ssl-certfile cert.pem --ssl-keyfile key.pem.
Explication :
Le pattern Saga gère les transactions cross-services sans 2PC. (1) Orchestration : un service central (Saga Orchestrator) envoie des commandes aux services et gère les compensations en cas d'échec. (2) Chorégraphie : les services publient des événements et réagissent aux événements des autres. En Python : Celery Canvas + Kafka/RabbitMQ. Chaque étape doit avoir une action de compensation (rollback métier).
Mentionner l'idempotence des messages : un message peut être reçu plusieurs fois (at-least-once delivery) → chaque handler doit être idempotent (même effet si exécuté N fois).
Explication :
TEMPLATES = [{'BACKEND': 'django.template.backends.django.DjangoTemplates',
'OPTIONS': {
'loaders': [('django.template.loaders.cached.Loader',
['django.template.loaders.filesystem.Loader',
'django.template.loaders.app_directories.Loader'])]
}}]Le
cached.Loader compile les templates une fois et les garde en mémoire. Désactiver en dev (DEBUG=True réactive le chargement à chaque requête).
Explication :
Pattern de schémas évolutif :class ArticleBase(BaseModel): titre: str; contenu: str
class ArticleCreate(ArticleBase): pass
class ArticleUpdate(BaseModel): # tout optionnel pour PATCH
titre: str | None = None
contenu: str | None = None
class ArticleResponse(ArticleBase):
id: int; created_at: datetime
model_config = ConfigDict(from_attributes=True)
Explication :
Deux syntaxes : (1) Ancienne :
class Meta: unique_together = [('auteur', 'titre')]. (2) Moderne (recommandée) : class Meta: constraints = [models.UniqueConstraint(fields=['auteur', 'titre'], name='unique_auteur_titre')]. La syntaxe UniqueConstraint permet des conditions (condition=Q(actif=True)) pour des contraintes partielles.
Explication :
Stack sécurisée : (1) K8s Secrets : base64 + RBAC pour l'accès. (2) Sealed Secrets ou External Secrets Operator : chiffrement fort, stockage Git sécurisé. (3) HashiCorp Vault : rotation automatique des secrets, audit log. L'app lit via
pydantic-settings depuis les variables d'environnement injectées par K8s. Ne jamais mettre de secret dans une image Docker.
En complément : utiliser imdsv2 sur AWS EC2, Workload Identity sur GKE pour éviter les clés statiques.
Explication :
from fastapi.responses import StreamingResponse
async def generate_csv():
yield 'nom,email\n'
async for user in get_users_stream():
yield f'{user.nom},{user.email}\n'
@app.get('/export/')
def export():
return StreamingResponse(generate_csv(), media_type='text/csv')Idéal pour les exports larges (CSV, JSON Lines) sans charger tout en mémoire.
Explication :
Stratégie de test des migrations : (1)
migrate --check en CI (fail si migration non appliquée). (2) Test de réversibilité : migrate app 0005 → migrate app 0004 (vérifier que database_backwards est implémenté). (3) Tester les data migrations séparément avec des fixtures. (4) squashmigrations testé en isolation. Plugin : pytest-django-migrations.
Explication :
@app.post('/transfer/')
async def transfer(data: TransferData, db: Session = Depends(get_db)):
with db.begin():
debit_account(db, data.from_id, data.amount)
credit_account(db, data.to_id, data.amount)
# Si une exception → rollback automatique
return {'status': 'ok'}Avec SQLAlchemy async :
async with db.begin():.
Explication :
Le Circuit Breaker empêche les appels répétés vers un service défaillant. États : Closed (normal) → Open (défaillant, rejette immédiatement) → Half-Open (test de récupération). import pybreaker
breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=30)
@breaker
async def call_external_service(): ...Retour rapide plutôt que timeout long.
Mentionner que la combinaison Circuit Breaker + Retry + Timeout est le pattern de résilience standard (Hystrix en Java, Polly en .NET, pybreaker/tenacity en Python).
Explication :
django-imagekit : from imagekit.models import ImageSpecField
from imagekit.processors import ResizeToFill
class Article(models.Model):
image = models.ImageField(upload_to='articles/')
thumbnail = ImageSpecField(source='image', processors=[ResizeToFill(300, 200)], format='JPEG', options={'quality': 85})Génère le thumbnail à la demande et le met en cache.
Explication :
Le Strangler Fig pattern est la stratégie recommandée : (1) Identifier les sous-domaines (notifications, rapports, search). (2) Extraire le moins couplé d'abord. (3) Mettre un API Gateway devant le monolithe. (4) Migrer progressivement les routes vers les nouveaux services FastAPI. (5) Le monolithe Django devient progressivement plus petit. Risque minimal vs big bang.
Mentionner les défis : données partagées (copie de tables, CDC avec Debezium), auth unifiée (token JWT partagé), monitoring distribué.