MFormations
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

SolutionUsageExemple
multiprocessingCPU-bound, tâches lourdesPool.map
asyncioI/O-bound, réseauaiohttp, asyncpg
C extensionsCalcul intensifnumpy, numba, Cython
JITOptimisation runtimePyPy
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èreFastAPIDjango + DRFDjango + Ninja
Performance25k req/s5k req/s15k req/s
Auto-docsOui (Swagger)Oui (DRF)Oui (Swagger)
ORMSQLAlchemy/TortoiseDjango ORMDjango ORM
Async natifOuiPartielOui
AdminNonOui (excellent)Non
Learning curveFaibleÉlevéeMoyenne
Use caseAPI microserviceFull-stackAPI 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

  1. Choix de structures de données : set > list pour membership (O(1) vs O(n))
  2. Compréhensions : [x*2 for x in lst] > list(map(...)) > boucle
  3. 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)
  1. f-strings > % > .format() > concatenation
  2. slots pour classes avec beaucoup d'instances
  3. @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)