Ce tutoriel couvre les fonctionnalités avancées de Python qui font la différence entre du code fonctionnel et du code professionnel : décorateurs, context managers, programmation asynchrone avec asyncio et typage statique.
Un décorateur est une fonction qui prend une fonction en entrée et retourne une nouvelle fonction enrichie.
Définition : Une fonction qui prend une autre fonction ou classe en entrée et retourne une version modifiée de celle-ci, ajoutant une nouvelle fonctionnalité sans modifier son code source.
But : Ajouter du comportement (logging, caching, validation, mesure de performance) à une fonction existante de manière réutilisable et élégante.
Pourquoi ici : Les décorateurs sont un élément clé de Python professionnel, particulièrement utilisés dans FastAPI, Flask et les frameworks modernes.
Définition : Une fonction interne qui capture et retient l'accès aux variables de sa fonction parente, même après que la fonction parente ait terminé son exécution.
But : Créer des fonctions personnalisées avec un contexte conservé, permettant aux décorateurs de maintenir l'état.
Pourquoi ici : Les closures sont le mécanisme fondamental derrière les décorateurs et les décorateurs paramétrés.
import functools
import time
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
logger = logging.getLogger(__name__)
# ── Décorateur de chronométrage ──
def timer(func):
@functools.wraps(func) # Préserve le nom et docstring de la fonction originale
def wrapper(*args, **kwargs): # *args=arguments positionnels, **kwargs=arguments nommés
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
logger.info(f"{func.__name__} exécuté en {elapsed:.4f}s")
return result
return wrapper
# ── Décorateur de retry avec backoff ──
def retry(max_attempts: int = 3, delay: float = 1.0, exceptions=(Exception,)):
"""Réessaye une fonction en cas d'échec."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts - 1:
raise
wait = delay * (2 ** attempt) # Backoff exponentiel
logger.warning(f"Tentative {attempt+1}/{max_attempts} échouée: {e}. Retry dans {wait}s")
time.sleep(wait)
return wrapper
return decorator
# ── Décorateur de cache simple (memoization) ──
def memoize(func):
@functools.wraps(func)
cache = {}
def wrapper(*args):
if args not in cache:
cache[args] = func(*args)
logger.info(f"Cache MISS pour {args}")
else:
logger.info(f"Cache HIT pour {args}")
return cache[args]
return wrapper
# ── Utilisation des décorateurs ──
@timer
@retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError,))
def fetch_data(url: str) -> dict:
"""Simule une requête HTTP avec retry automatique."""
import random
if random.random() < 0.4: # 40% de chance d'échec
raise ConnectionError(f"Connexion impossible à {url}")
return {"url": url, "data": "..."}
@memoize
def fibonacci(n: int) -> int:
if n <= 1: return n
return fibonacci(n - 1) + fibonacci(n - 2)
# Test
result = fetch_data("https://api.example.com")
print(fibonacci(10)) # 55 — calculé une fois, puis depuis le cache__name__ == 'wrapper' au lieu de son vrai nom — mauvais pour le debugging et la documentation automatique. *args capture tous les arguments positionnels, **kwargs tous les arguments nommés, rendant le décorateur compatible avec n'importe quelle signature de fonction.
Définition : Un objet Python qui implémente les méthodes __enter__ et __exit__ pour gérer l'acquisition et la libération de ressources de manière garantie.
But : Assurer qu'une ressource (fichier, connexion DB, verrou) est proprement libérée même en cas d'erreur, sans code de nettoyage manuel répétitif.
Pourquoi ici : Utilisé partout : with open() pour les fichiers, with db_session() pour les BDD, with asyncio.lock() pour la synchronisation.
Définition : Deux méthodes spéciales : __enter__ (s'exécute au début du bloc with) et __exit__ (s'exécute à la fin, même en cas d'exception).
But : __enter__ acquiert la ressource et la retourne pour utilisation. __exit__ libère la ressource et gère les exceptions (peut supprimer l'exception si elle retourne True).
Pourquoi ici : Permet une utilisation élégante : with resource() as r: ... au lieu de try/finally manuel.
from contextlib import contextmanager
import sqlite3
import time
# ── Context manager avec @contextmanager ──
@contextmanager
def database_connection(db_path: str):
"""Ouvre et ferme une connexion DB automatiquement."""
conn = None
try:
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row
print(f"✅ Connexion à {db_path} ouverte")
yield conn # Point d'entrée du bloc "with"
conn.commit()
print("✅ Transaction validée (commit)")
except Exception as e:
if conn: conn.rollback()
print(f"❌ Erreur — rollback effectué : {e}")
raise
finally:
if conn: conn.close()
print("🔒 Connexion fermée")
# ── Context manager comme classe ──
class Timer:
"""Chronomètre utilisable comme context manager."""
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, *args):
self.elapsed = time.perf_counter() - self.start
print(f"⏱ Durée : {self.elapsed:.4f}s")
# ── Utilisation ──
with database_connection(":memory:") as conn:
conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
conn.execute("INSERT INTO users (name) VALUES (?)", ("Alderi",))
users = conn.execute("SELECT * FROM users").fetchall()
print(f"Utilisateurs : {[dict(u) for u in users]}")
# ← La connexion est fermée automatiquement ici
with Timer() as t:
import time; time.sleep(0.1)
print(f"Elapsed: {t.elapsed:.4f}s")yield : tout avant yield = __enter__, tout après = __exit__. C'est plus lisible qu'une classe pour les cas simples. La version classe explicite montre les méthodes __enter__ (retourne self) et __exit__ (prend 3 paramètres d'exception, ignorés ici). Même avec une erreur SQL, la connexion se ferme grâce au bloc finally implicite.
try/finally partout. Une ressource oubliée = fuite mémoire ou lock qui se bloque. Les context managers rendent impossible d'oublier le nettoyage.
Définition : Une fonction définie avec async def qui peut être interrompue avec await et reprendre plus tard, sans bloquer le thread.
But : Permettre l'exécution de milliers d'opérations I/O (réseau, fichiers) de manière efficace sur un seul thread, en ne laissant l'exécution que quand c'est nécessaire d'attendre.
Pourquoi ici : Essential pour les APIs web, les scrapers, les clients WebSocket modernes.
Définition : Bibliothèque Python standard pour la programmation asynchrone, fournissant une event loop qui gère les coroutines.
But : Fournir les outils (gather, Queue, sleep, run) pour écrire du code asynchrone sans gérer manuellement les threads.
Pourquoi ici : Standard de facto pour le async/await en Python, remplace Twisted et Gevent.
Définition : La boucle centrale de asyncio qui exécute les coroutines, vérifie lesquelles sont prêtes à reprendre après une opération I/O.
But : Dispatcher l'exécution entre plusieurs coroutines efficacement, maximisant l'utilisation du CPU pendant que d'autres attendent.
Pourquoi ici : asyncio.run(main()) crée et exécute la boucle d'événements—c'est le point d'entrée.
import asyncio
import aiohttp
import time
# ── Requêtes HTTP asynchrones ──
async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict: # async def = définit une coroutine
"""Télécharge une URL de manière asynchrone."""
async with session.get(url) as response: # await implicite ici
data = await response.json() # await = libère la boucle d'événements, attend le résultat
print(f"✅ {url} → {response.status}")
return data
async def fetch_multiple(urls: list[str]) -> list[dict]:
"""Télécharge plusieurs URLs EN PARALLÈLE."""
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
# asyncio.gather exécute toutes les coroutines en parallèle (pas séquentiel)
results = await asyncio.gather(*tasks, return_exceptions=True) # gather = lance tout et attend tous
return [r for r in results if not isinstance(r, Exception)]
# ── Comparaison sync vs async ──
URLS = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3",
"https://jsonplaceholder.typicode.com/posts/4",
"https://jsonplaceholder.typicode.com/posts/5",
]
async def main():
start = time.perf_counter()
results = await fetch_multiple(URLS)
elapsed = time.perf_counter() - start
print(f"\n{len(results)} requêtes parallèles en {elapsed:.2f}s")
# vs ~5s en séquentiel (1s par requête × 5)
asyncio.run(main())
# ── asyncio.Queue — producteur / consommateur ──
# Queue permet une communication sûre entre coroutines asynchrones
async def producer(queue: asyncio.Queue, items: list):
for item in items:
await queue.put(item)
await asyncio.sleep(0.1) # Simule la production
await queue.put(None) # Signal de fin
async def consumer(queue: asyncio.Queue):
while True:
item = await queue.get()
if item is None: break
print(f"Traitement : {item}")
await asyncio.sleep(0.05) # Simule le traitement
queue.task_done()async def = coroutine, await = point de pause. Sans await, la coroutine ne s'exécute pas, elle est juste créée. asyncio.gather lance les 5 requêtes HTTP en parallèle (elles s'interruptent mutuellement en attendant le réseau) vs une boucle séquentielle qui attendrait 5 secondes totales. Queue permet au producteur d'ajouter des items sans bloquer et au consommateur de les récupérer à son rythme—essentiellement un buffer thread-safe mais pour les coroutines.
Définition : Une annotation de type dans la signature d'une fonction ou variable (ex: def func(x: int) -> str:) indiquant le type attendu.
But : Aider les outils de vérification statique (mypy) à détecter les erreurs sans exécuter le code, et documenter les attentes.
Pourquoi ici : Python est typé dynamiquement, mais les type hints ajoutent la sécurité du typage statique en opt-in—recommandé en code professionnel.
Définition : TypeVar('T') crée une variable de type générique. Generic[T] permet à une classe de paramétriser sur ce type.
But : Écrire des classes et fonctions réutilisables qui travaillent sur n'importe quel type en conservant la vérification de type. Exemple : Result[User] vs Result[str].
Pourquoi ici : Les generics sont essentiels pour les conteneurs typés et les patterns fonctionnels (Result, Option, Either).
from dataclasses import dataclass, field
from typing import Optional, Union, TypeVar, Generic, Protocol
from datetime import datetime
from enum import Enum
# ── Enum typé ──
class Status(Enum):
PENDING = "pending"
ACTIVE = "active"
INACTIVE = "inactive"
# ── Dataclass avec validation ──
@dataclass
class User:
id: int
name: str
email: str
status: Status = Status.ACTIVE
created_at: datetime = field(default_factory=datetime.now)
tags: list[str] = field(default_factory=list)
def __post_init__(self):
# Validation à la construction
if "@" not in self.email:
raise ValueError(f"Email invalide : {self.email}")
self.name = self.name.strip()
@property
def display_name(self) -> str:
return f"{self.name} <{self.email}>"
# ── Generics — classes et fonctions typées réutilisables ──
T = TypeVar('T') # T = variable de type générique
@dataclass
class Result(Generic[T]): # Rend Result typé : Result[User], Result[int]
"""Encapsule un résultat ou une erreur (pattern Result/Either)."""
success: bool
data: Optional[T] = None
error: Optional[str] = None
@classmethod
def ok(cls, data: T) -> "Result[T]":
return cls(success=True, data=data)
@classmethod
def fail(cls, error: str) -> "Result[T]":
return cls(success=False, error=error)
def find_user(user_id: int, users: list[User]) -> Result[User]:
for user in users:
if user.id == user_id:
return Result.ok(user)
return Result.fail(f"Utilisateur {user_id} introuvable")
# ── Test ──
users = [User(1, "Alderi", "alderi@example.com")]
result = find_user(1, users)
if result.success:
print(result.data.display_name) # Alderi <alderi@example.com>
result2 = find_user(999, users)
print(result2.error) # Utilisateur 999 introuvablefind_user("hello", users) serait flaggé comme erreur type). Les generics permettent Result[User] : le checker sait que result.data est un User, pas un Any—les IDE peuvent donc proposer l'auto-complétion pour ses méthodes.
func(None) ne sera découvert qu'à la runtime avec une erreur. Avec mypy, c'est détecté avant de pousser le code. Aucun coût de runtime, aucun changement de syntaxe complexe—juste de l'information pour les outils.
pip install mypy && mypy mon_fichier.py