Gestion des erreurs en Python
Construisez des applications Python robustes avec une validation appropriée des entrées, des exceptions significatives et une gestion élégante des défaillances. Une bonne gestion des erreurs simplifie le débogage et rend les systèmes plus fiables.
Quand utiliser cette compétence
- Valider les entrées utilisateur et les paramètres API
- Concevoir des hiérarchies d'exceptions pour les applications
- Gérer les défaillances partielles dans les opérations par lot
- Convertir les données externes en types de domaine
- Construire des messages d'erreur conviviaux
- Implémenter des modèles de validation fail-fast
Concepts fondamentaux
1. Fail Fast
Validez les entrées tôt, avant les opérations coûteuses. Signalez tous les erreurs de validation à la fois quand c'est possible.
2. Exceptions significatives
Utilisez les types d'exceptions appropriés avec contexte. Les messages doivent expliquer ce qui a échoué, pourquoi, et comment le corriger.
3. Défaillances partielles
Dans les opérations par lot, ne laissez pas une défaillance tout annuler. Suivez séparément les succès et les défaillances.
4. Préserver le contexte
Chaînez les exceptions pour maintenir la piste d'erreur complète pour le débogage.
Démarrage rapide
def fetch_page(url: str, page_size: int) -> Page:
if not url:
raise ValueError("'url' is required")
if not 1 <= page_size <= 100:
raise ValueError(f"'page_size' must be 1-100, got {page_size}")
# Now safe to proceed...
Modèles fondamentaux
Modèle 1 : Validation précoce des entrées
Validez toutes les entrées aux limites de l'API avant tout traitement.
def process_order(
order_id: str,
quantity: int,
discount_percent: float,
) -> OrderResult:
"""Process an order with validation."""
# Validate required fields
if not order_id:
raise ValueError("'order_id' is required")
# Validate ranges
if quantity <= 0:
raise ValueError(f"'quantity' must be positive, got {quantity}")
if not 0 <= discount_percent <= 100:
raise ValueError(
f"'discount_percent' must be 0-100, got {discount_percent}"
)
# Validation passed, proceed with processing
return _process_validated_order(order_id, quantity, discount_percent)
Modèle 2 : Convertir en types de domaine tôt
Analysez les chaînes et les données externes en objets de domaine typés aux limites du système.
from enum import Enum
class OutputFormat(Enum):
JSON = "json"
CSV = "csv"
PARQUET = "parquet"
def parse_output_format(value: str) -> OutputFormat:
"""Parse string to OutputFormat enum.
Args:
value: Format string from user input.
Returns:
Validated OutputFormat enum member.
Raises:
ValueError: If format is not recognized.
"""
try:
return OutputFormat(value.lower())
except ValueError:
valid_formats = [f.value for f in OutputFormat]
raise ValueError(
f"Invalid format '{value}'. "
f"Valid options: {', '.join(valid_formats)}"
)
# Usage at API boundary
def export_data(data: list[dict], format_str: str) -> bytes:
output_format = parse_output_format(format_str) # Fail fast
# Rest of function uses typed OutputFormat
...
Modèle 3 : Pydantic pour la validation complexe
Utilisez les modèles Pydantic pour la validation d'entrée structurée avec messages d'erreur automatiques.
from pydantic import BaseModel, Field, field_validator
class CreateUserInput(BaseModel):
"""Input model for user creation."""
email: str = Field(..., min_length=5, max_length=255)
name: str = Field(..., min_length=1, max_length=100)
age: int = Field(ge=0, le=150)
@field_validator("email")
@classmethod
def validate_email_format(cls, v: str) -> str:
if "@" not in v or "." not in v.split("@")[-1]:
raise ValueError("Invalid email format")
return v.lower()
@field_validator("name")
@classmethod
def normalize_name(cls, v: str) -> str:
return v.strip().title()
# Usage
try:
user_input = CreateUserInput(
email="user@example.com",
name="john doe",
age=25,
)
except ValidationError as e:
# Pydantic provides detailed error information
print(e.errors())
Modèle 4 : Mapper les erreurs aux exceptions standard
Utilisez les types d'exceptions intégrés de Python de manière appropriée, en ajoutant du contexte selon les besoins.
| Type de défaillance | Exception | Exemple |
|---|---|---|
| Entrée invalide | ValueError |
Valeurs de paramètres incorrectes |
| Mauvais type | TypeError |
Chaîne attendue, int reçu |
| Élément manquant | KeyError |
Clé dict non trouvée |
| Défaillance opérationnelle | RuntimeError |
Service indisponible |
| Délai d'attente dépassé | TimeoutError |
L'opération a pris trop de temps |
| Fichier non trouvé | FileNotFoundError |
Le chemin n'existe pas |
| Accès refusé | PermissionError |
Accès interdit |
# Good: Specific exception with context
raise ValueError(f"'page_size' must be 1-100, got {page_size}")
# Avoid: Generic exception, no context
raise Exception("Invalid parameter")
Modèles avancés
Modèle 5 : Exceptions personnalisées avec contexte
Créez des exceptions spécifiques au domaine qui transportent des informations structurées.
class ApiError(Exception):
"""Base exception for API errors."""
def __init__(
self,
message: str,
status_code: int,
response_body: str | None = None,
) -> None:
self.status_code = status_code
self.response_body = response_body
super().__init__(message)
class RateLimitError(ApiError):
"""Raised when rate limit is exceeded."""
def __init__(self, retry_after: int) -> None:
self.retry_after = retry_after
super().__init__(
f"Rate limit exceeded. Retry after {retry_after}s",
status_code=429,
)
# Usage
def handle_response(response: Response) -> dict:
match response.status_code:
case 200:
return response.json()
case 401:
raise ApiError("Invalid credentials", 401)
case 404:
raise ApiError(f"Resource not found: {response.url}", 404)
case 429:
retry_after = int(response.headers.get("Retry-After", 60))
raise RateLimitError(retry_after)
case code if 400 <= code < 500:
raise ApiError(f"Client error: {response.text}", code)
case code if code >= 500:
raise ApiError(f"Server error: {response.text}", code)
Modèle 6 : Chaînage d'exceptions
Préservez l'exception d'origine lors du relancement pour maintenir la piste de débogage.
import httpx
class ServiceError(Exception):
"""High-level service operation failed."""
pass
def upload_file(path: str) -> str:
"""Upload file and return URL."""
try:
with open(path, "rb") as f:
response = httpx.post("https://upload.example.com", files={"file": f})
response.raise_for_status()
return response.json()["url"]
except FileNotFoundError as e:
raise ServiceError(f"Upload failed: file not found at '{path}'") from e
except httpx.HTTPStatusError as e:
raise ServiceError(
f"Upload failed: server returned {e.response.status_code}"
) from e
except httpx.RequestError as e:
raise ServiceError(f"Upload failed: network error") from e
Modèle 7 : Traitement par lot avec défaillances partielles
Ne laissez jamais un élément défectueux annuler un lot entier. Suivez les résultats par élément.
from dataclasses import dataclass
@dataclass
class BatchResult[T]:
"""Results from batch processing."""
succeeded: dict[int, T] # index -> result
failed: dict[int, Exception] # index -> error
@property
def success_count(self) -> int:
return len(self.succeeded)
@property
def failure_count(self) -> int:
return len(self.failed)
@property
def all_succeeded(self) -> bool:
return len(self.failed) == 0
def process_batch(items: list[Item]) -> BatchResult[ProcessedItem]:
"""Process items, capturing individual failures.
Args:
items: Items to process.
Returns:
BatchResult with succeeded and failed items by index.
"""
succeeded: dict[int, ProcessedItem] = {}
failed: dict[int, Exception] = {}
for idx, item in enumerate(items):
try:
result = process_single_item(item)
succeeded[idx] = result
except Exception as e:
failed[idx] = e
return BatchResult(succeeded=succeeded, failed=failed)
# Caller handles partial results
result = process_batch(items)
if not result.all_succeeded:
logger.warning(
f"Batch completed with {result.failure_count} failures",
failed_indices=list(result.failed.keys()),
)
Modèle 8 : Signalement de progression pour les opérations longues
Fournissez une visibilité sur la progression par lot sans couplage de la logique métier à l'interface.
from collections.abc import Callable
ProgressCallback = Callable[[int, int, str], None] # current, total, status
def process_large_batch(
items: list[Item],
on_progress: ProgressCallback | None = None,
) -> BatchResult:
"""Process batch with optional progress reporting.
Args:
items: Items to process.
on_progress: Optional callback receiving (current, total, status).
"""
total = len(items)
succeeded = {}
failed = {}
for idx, item in enumerate(items):
if on_progress:
on_progress(idx, total, f"Processing {item.id}")
try:
succeeded[idx] = process_single_item(item)
except Exception as e:
failed[idx] = e
if on_progress:
on_progress(total, total, "Complete")
return BatchResult(succeeded=succeeded, failed=failed)
Résumé des bonnes pratiques
- Validez tôt - Vérifiez les entrées avant les opérations coûteuses
- Utilisez des exceptions spécifiques -
ValueError,TypeError, pasExceptiongénérique - Incluez du contexte - Les messages doivent expliquer quoi, pourquoi et comment corriger
- Convertissez les types aux limites - Analysez les chaînes en énums/types de domaine tôt
- Chaînez les exceptions - Utilisez
raise ... from epour préserver les info de débogage - Gérez les défaillances partielles - N'annulez pas les lots sur des erreurs d'un seul élément
- Utilisez Pydantic - Pour la validation d'entrée complexe avec erreurs structurées
- Documentez les modes de défaillance - Les docstrings doivent lister les exceptions possibles
- Loggez avec contexte - Incluez des IDs, des comptages et d'autres infos de débogage
- Testez les chemins d'erreur - Vérifiez que les exceptions sont levées correctement