Build robust Python applications with proper input validation, meaningful exceptions, and graceful failure handling. Good error handling makes debugging easier and systems more reliable. - Validating user input and API parameters - Designing exception hierarchies for applications
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...
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)
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 ...
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())
ValueErrorTypeErrorKeyErrorRuntimeErrorTimeoutErrorFileNotFoundErrorPermissionError# 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")
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)
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
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()), )
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)
ValueError, TypeError, not generic Exceptionraise ... from e to preserve debug info