Modern Backend Engineering
Chapitre 3
Chapitre 03 — Python
Chapitre 03 — Python
Cours complet — Python
1. GIL (Global Interpreter Lock)
Qu'est-ce que le GIL ?
- Mutex qui protège l'interpréteur CPython
- Un seul thread peut exécuter du bytecode à la fois
- Introduit pour simplifier la gestion mémoire (ref counting threadsafe)
Impact sur les performances
CPU-bound : GIL bloque le parallélisme → pas de gain multi-thread
I/O-bound : GIL libéré pendant les appels système → multi-thread efficace
# CPU-bound — GIL limitant
def cpu_intensive(n):
"""Calcul inutilement lourd — GIL va limiter"""
return sum(i * i for i in range(n))
# I/O-bound — GIL libéré
def io_intensive():
"""Appel réseau — GIL libéré pendant l'attente"""
import requests
return requests.get("https://api.example.com").json()
Solutions pour contourner le GIL
| Solution | Usage | Exemple |
|---|---|---|
| multiprocessing | CPU-bound, tâches lourdes | Pool.map |
| asyncio | I/O-bound, réseau | aiohttp, asyncpg |
| C extensions | Calcul intensif | numpy, numba, Cython |
| JIT | Optimisation runtime | PyPy |
| Subinterpreters (PEP 554) | Nouveau (Python 3.12+) | Per-interpreter GIL |
from multiprocessing import Pool
def process_chunk(chunk):
return [x * x for x in chunk]
with Pool(4) as p:
results = p.map(process_chunk, data_chunks)
PEP 703 — no-GIL (Python 3.13+ experimental)
- GIL optionnel via
--disable-gilà la compilation - Nouveau système de référence counting (biased reference counting)
- Compatible avec l'extension C API (avec modifications)
- Performance : 10-20% de perte pour single-thread, gain pour multi-thread
2. async/await (asyncio)
Event loop asyncio
import asyncio
async def fetch_data(url: str) -> dict:
"""Coroutine asynchrone — ne bloque pas l'event loop"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
return await resp.json()
async def main():
# Concurrence : 3 tâches en parallèle
tasks = [
fetch_data("https://api.example.com/1"),
fetch_data("https://api.example.com/2"),
fetch_data("https://api.example.com/3"),
]
results = await asyncio.gather(*tasks)
return results
# Lancer
asyncio.run(main())
Coroutines vs Tasks vs Futures
- Coroutine : fonction async def → awaitable
- Task : coroutine enveloppée dans asyncio.create_task()
- Future : valeur future (comme Promise en JS)
async def main():
# Créer une Task (planifiée dans l'event loop)
task = asyncio.create_task(fetch_data("https://..."))
# Attendre le résultat
result = await task
# Future (bas niveau)
future = asyncio.get_event_loop().create_future()
future.set_result("done")
await future
Async context managers et iterators
# Async context manager
class DatabaseConnection:
async def __aenter__(self):
self.conn = await connect()
return self.conn
async def __aexit__(self, *args):
await self.conn.close()
# Async iterator
class AsyncRange:
def __init__(self, n):
self.n = n
self.i = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.i < self.n:
await asyncio.sleep(0.1) # Simule I/O
self.i += 1
return self.i
raise StopAsyncIteration
3. FastAPI vs Django
FastAPI
- Basé sur : Starlette (ASGI) + Pydantic
- Performance : ~25k req/s (Uvicorn)
- Auto-docs : Swagger + ReDoc (OpenAPI)
- Validation : Pydantic schemas
- Dépendance injection : FastAPI.Depends()
- Background tasks : BackgroundTasks
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks
from pydantic import BaseModel, EmailStr
from typing import Annotated
app = FastAPI(title="Modern API", version="1.0.0")
# Schemas
class UserCreate(BaseModel):
name: str = Field(min_length=2, max_length=100)
email: EmailStr
age: int = Field(ge=0, le=150)
class UserResponse(BaseModel):
id: int
name: str
email: str
created_at: datetime
# Dépendance
async def get_db():
async with Database() as db:
yield db
# Route
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(
data: UserCreate,
db: Annotated[Database, Depends(get_db)],
tasks: BackgroundTasks,
):
user = await db.create_user(data)
tasks.add_task(send_welcome_email, user.email)
return user
Django
- Batteries included : ORM, admin, auth, forms, migrations
- DRF (Django REST Framework) : API REST
- Ninja : Alternative DRF avec Pydantic + performance
- ASGI support depuis Django 3.0
- ORM : mature, migrations, relations complexes
# models.py
from django.db import models
class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
indexes = [
models.Index(fields=['email']),
]
# serializers.py (DRF)
from rest_framework import serializers
class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['id', 'name', 'email', 'created_at']
# views.py (Ninja — moderne)
from ninja import NinjaAPI, ModelSchema
api = NinjaAPI()
class UserSchema(ModelSchema):
class Meta:
model = User
fields = ['id', 'name', 'email']
@api.post("/users", response=UserSchema)
def create(request, data: UserSchema):
return User.objects.create(**data.dict())
Comparaison
| Critère | FastAPI | Django + DRF | Django + Ninja |
|---|---|---|---|
| Performance | 25k req/s | 5k req/s | 15k req/s |
| Auto-docs | Oui (Swagger) | Oui (DRF) | Oui (Swagger) |
| ORM | SQLAlchemy/Tortoise | Django ORM | Django ORM |
| Async natif | Oui | Partiel | Oui |
| Admin | Non | Oui (excellent) | Non |
| Learning curve | Faible | Élevée | Moyenne |
| Use case | API microservice | Full-stack | API Django |
4. Typing (Pydantic, mypy)
Type hints avancés (Python 3.12+)
from typing import (
assert_never,
Literal,
TypedDict,
Never,
Self,
overload,
Concatenate,
ParamSpec,
TypeVar,
Generic,
)
from typing_extensions import override, @deprecated
# TypeVar bounds
T = TypeVar('T', bound=BaseModel)
# ParamSpec (callable generics)
P = ParamSpec('P')
R = TypeVar('R')
def timed(func: Callable[P, R]) -> Callable[P, R]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
start = time()
result = func(*args, **kwargs)
log(f"{func.__name__} took {time() - start:.3f}s")
return result
return wrapper
# TypedDict
class UserDict(TypedDict):
id: int
name: str
email: NotRequired[str] # Python 3.11+
# Literal types
def process_status(status: Literal["active", "inactive", "pending"]) -> str: ...
Pydantic
from pydantic import BaseModel, Field, ConfigDict, model_validator, field_validator
from datetime import datetime
from typing import Optional
class UserBase(BaseModel):
model_config = ConfigDict(from_attributes=True, extra="forbid")
name: str = Field(min_length=2, max_length=100)
email: str = Field(pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$")
age: Optional[int] = Field(default=None, ge=0, le=150)
@field_validator("name")
@classmethod
def name_must_be_proper(cls, v: str) -> str:
return v.strip().title()
@model_validator(mode="after")
def check_something(self) -> Self:
if self.age and self.age < 18 and "admin" in self.name.lower():
raise ValueError("Admin must be 18+")
return self
class UserCreate(UserBase):
pass
class UserResponse(UserBase):
id: int
created_at: datetime
5. Performance Python
Profiling
# cProfile
python -m cProfile -o profile.stats my_script.py
# py-spy (sampling profiler, sans modification du code)
py-spy record -o profile.svg -- python my_script.py
# Scalene (CPU + GPU + memory)
pip install scalene
scalene my_script.py
Optimisations
- Choix de structures de données : set > list pour membership (O(1) vs O(n))
- Compréhensions :
[x*2 for x in lst]>list(map(...))> boucle - Local variable binding :
# Lent
def slow(items):
for item in items:
math.sin(item)
# Rapide (bind local)
def fast(items):
sin = math.sin
for item in items:
sin(item)
- f-strings > % > .format() > concatenation
- slots pour classes avec beaucoup d'instances
- @lru_cache / @cache pour fonctions pures
Python vs C extensions
# Pure Python
def sum_array(arr):
total = 0
for x in arr:
total += x
return total
# numpy (C)
import numpy as np
result = np.sum(arr) # 100x plus rapide
# numba (JIT)
from numba import njit
@njit
def sum_numba(arr):
total = 0
for x in arr:
total += x
return total
6. Packaging (pip, poetry, uv)
Évolution des outils
pip + virtualenv → Traditionnel
Pipenv → 2017-2020 (déclin)
Poetry → 2018-2025 (standard)
PDM → 2021+ (PEP 582)
uv → 2024+ (Rust, ultra-rapide)
Poetry
poetry new my-project
poetry add fastapi uvicorn[standard]
poetry add --dev pytest mypy ruff
poetry run python main.py
# pyproject.toml
[tool.poetry]
name = "my-project"
version = "0.1.0"
python = "^3.12"
[tool.poetry.dependencies]
fastapi = "^0.115"
uvicorn = {extras = ["standard"], version = "^0.30"}
asyncpg = "^0.29"
pydantic = "^2.8"
redis = "^5.0"
[tool.poetry.group.dev.dependencies]
pytest = "^8.0"
pytest-asyncio = "^0.24"
mypy = "^1.11"
ruff = "^0.6"
uv (ultra-rapide, Rust)
# 10-100x plus rapide que pip/poetry
uv pip install fastapi
uv sync
uv run python main.py
7. Patterns Python
Repository Pattern
from abc import ABC, abstractmethod
from typing import Generic, TypeVar
T = TypeVar('T', bound=BaseModel)
class Repository(ABC, Generic[T]):
@abstractmethod
async def get(self, id: int) -> T | None: ...
@abstractmethod
async def list(self, skip: int = 0, limit: int = 100) -> list[T]: ...
@abstractmethod
async def create(self, data: T) -> T: ...
@abstractmethod
async def update(self, id: int, data: T) -> T: ...
@abstractmethod
async def delete(self, id: int) -> bool: ...
class PostgresUserRepository(Repository[User]):
def __init__(self, session: AsyncSession):
self.session = session
async def get(self, id: int) -> User | None:
return await self.session.get(User, id)
async def create(self, data: UserCreate) -> User:
user = User(**data.model_dump())
self.session.add(user)
await self.session.commit()
await self.session.refresh(user)
return user
Service Layer
class UserService:
def __init__(self, repo: Repository[User], cache: CacheService):
self.repo = repo
self.cache = cache
async def get_user(self, user_id: int) -> UserResponse:
# Cache-aside
cached = await self.cache.get(f"user:{user_id}")
if cached:
return UserResponse(**cached)
user = await self.repo.get(user_id)
if not user:
raise HTTPException(status_code=404)
await self.cache.set(f"user:{user_id}", user.model_dump(), ttl=300)
return UserResponse.model_validate(user)
Dependency Injection (FastAPI)
from fastapi import Depends
from typing import Annotated
# Providers
async def get_session() -> AsyncSession:
async with async_session() as session:
yield session
async def get_user_repo(session: Annotated[AsyncSession, Depends(get_session)]) -> Repository[User]:
return PostgresUserRepository(session)
async def get_user_service(repo: Annotated[Repository[User], Depends(get_user_repo)]) -> UserService:
return UserService(repo, RedisCache())
# Routes
@app.get("/users/{user_id}")
async def get_user(
user_id: int,
service: Annotated[UserService, Depends(get_user_service)],
):
return await service.get_user(user_id)
Références
- CPython internals (python.org)
- FastAPI documentation (fastapi.tiangolo.com)
- Pydantic documentation (docs.pydantic.dev)
- asyncio official docs
- Django documentation (docs.djangoproject.com)