python-error-handling by wshobson/agents
npx skills add https://github.com/wshobson/agents --skill python-error-handling通过适当的输入验证、有意义的异常处理和优雅的故障处理,构建健壮的 Python 应用程序。良好的错误处理使调试更轻松,系统更可靠。
在昂贵的操作之前尽早验证输入。尽可能一次性报告所有验证错误。
使用带有上下文的适当异常类型。错误信息应解释什么失败了、为什么失败以及如何修复。
在批量操作中,不要让一个故障中止所有操作。分别跟踪成功和失败。
链接异常以维护完整的错误追踪路径,便于调试。
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...
在任何处理开始之前,在 API 边界验证所有输入。
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)
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
在系统边界将字符串和外部数据解析为类型化的领域对象。
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
...
使用 Pydantic 模型进行结构化输入验证,并自动生成错误信息。
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())
适当使用 Python 的内置异常类型,并根据需要添加上下文。
| 故障类型 | 异常 | 示例 |
|---|---|---|
| 无效输入 | ValueError | 错误的参数值 |
| 类型错误 | TypeError | 期望字符串,得到整数 |
| 缺少项 | KeyError | 字典键未找到 |
| 操作故障 | RuntimeError | 服务不可用 |
| 超时 | TimeoutError | 操作耗时过长 |
| 文件未找到 | FileNotFoundError | 路径不存在 |
| 权限被拒绝 | PermissionError | 访问被禁止 |
# 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()),
)
在不将业务逻辑与 UI 耦合的情况下,提供批量进度的可见性。
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,而非通用的 Exceptionraise ... from e 保留调试信息每周安装量
3.5K
仓库
GitHub 星标
32.3K
首次出现
2026年1月30日
安全审计
安装于
opencode2.8K
gemini-cli2.8K
codex2.8K
claude-code2.5K
github-copilot2.5K
cursor2.5K
Build robust Python applications with proper input validation, meaningful exceptions, and graceful failure handling. Good error handling makes debugging easier and systems more reliable.
Validate inputs early, before expensive operations. Report all validation errors at once when possible.
Use appropriate exception types with context. Messages should explain what failed, why, and how to fix it.
In batch operations, don't let one failure abort everything. Track successes and failures separately.
Chain exceptions to maintain the full error trail for debugging.
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...
Validate all inputs at API boundaries before any processing begins.
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)
Parse strings and external data into typed domain objects at system boundaries.
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
...
Use Pydantic models for structured input validation with automatic error messages.
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())
Use Python's built-in exception types appropriately, adding context as needed.
| Failure Type | Exception | Example |
|---|---|---|
| Invalid input | ValueError | Bad parameter values |
| Wrong type | TypeError | Expected string, got int |
| Missing item | KeyError | Dict key not found |
| Operational failure | RuntimeError | Service unavailable |
| Timeout | TimeoutError | Operation took too long |
# 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")
Create domain-specific exceptions that carry structured information.
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)
Preserve the original exception when re-raising to maintain the debug trail.
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
Never let one bad item abort an entire batch. Track results per item.
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()),
)
Provide visibility into batch progress without coupling business logic to UI.
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 infoWeekly Installs
3.5K
Repository
GitHub Stars
32.3K
First Seen
Jan 30, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykPass
Installed on
opencode2.8K
gemini-cli2.8K
codex2.8K
claude-code2.5K
github-copilot2.5K
cursor2.5K
React 组合模式指南:Vercel 组件架构最佳实践,提升代码可维护性
102,200 周安装
| File not found | FileNotFoundError | Path doesn't exist |
| Permission denied | PermissionError | Access forbidden |