Loading...
Loading...
Python error handling patterns including input validation, exception hierarchies, and partial failure handling. Use when implementing validation logic, designing exception strategies, handling batch processing failures, or building robust APIs.
npx skill4agent add wshobson/agents python-error-handlingdef 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())| Failure Type | Exception | Example |
|---|---|---|
| Invalid input | | Bad parameter values |
| Wrong type | | Expected string, got int |
| Missing item | | Dict key not found |
| Operational failure | | Service unavailable |
| Timeout | | Operation took too long |
| File not found | | Path doesn't exist |
| Permission denied | | Access forbidden |
# 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 efrom 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)ValueErrorTypeErrorExceptionraise ... from e