python-error-handling

Python 错误处理

通过恰当的输入验证、有意义的异常和优雅的失败处理，构建稳健的 Python 应用程序。良好的错误处理能让调试更轻松，让系统更可靠。

适用场景

• 验证用户输入和 API 参数
• 设计应用程序的异常层级
• 处理批量操作中的部分失败
• 将外部数据转换为领域类型
• 构建对用户友好的错误消息
• 实现快速失败验证模式

核心概念

1. 快速失败

在进行耗时操作之前尽早验证输入。尽可能一次性报告所有验证错误。

2. 有意义的异常

使用恰当的异常类型并附上上下文。错误消息应说明哪里失败、原因是什么，以及如何修复。

3. 部分失败

在批量操作中，不要让单次失败中止所有操作。分别跟踪成功项和失败项。

4. 保留上下文

链接异常，以便在调试时保留完整的错误链路。

快速入门

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...

基础模式

模式 1：尽早验证输入

在 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)

模式 2：尽早转换为领域类型

在系统边界处将字符串和外部数据解析为带类型的领域对象。

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
    ...

模式 3：使用 Pydantic 进行复杂验证

使用 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="[email protected]",
        name="john doe",
        age=25,
    )
except ValidationError as e:
    # Pydantic provides detailed error information
    print(e.errors())

模式 4：将错误映射为标准异常

恰当使用 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")

高级模式

模式 5：带上下文的自定义异常

创建携带结构化信息的领域专用异常。

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)

模式 6：异常链

重新引发异常时保留原始异常，以维持完整的调试线索。

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

模式 7：支持部分失败的批量处理

不要让单个错误项中止整个批次。逐项跟踪处理结果。

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()),
    )

模式 8：长时间操作的进度报告

显示批量处理进度，同时避免业务逻辑与 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)

最佳实践总结

1. 尽早验证 - 在进行耗时操作之前检查输入
2. 使用具体的异常 - 使用 ValueError、TypeError，而不是通用的 Exception
3. 包含上下文 - 消息应说明是什么、为什么，以及如何修复
4. 在边界处转换类型 - 尽早将字符串解析为枚举或领域类型
5. 链接异常 - 使用 raise ... from e 保留调试信息
6. 处理部分失败 - 不要因单个项目出错而中止整个批次
7. 使用 Pydantic - 对复杂输入进行验证并生成结构化错误
8. 记录失败模式 - 文档字符串应列出可能出现的异常
9. 记录带上下文的日志 - 包含 ID、数量及其他调试信息
10. 测试错误路径 - 验证异常是否按预期引发

---

易踩点

• raise X from None 会抑制原因链；raise X 会保留 __context__ — 堆栈跟踪不同。当较低层的异常无关紧要时，使用 from None。
• 在 except 中使用 raise（裸 raise）会重新抛出原始异常；raise e 会用新的回溯重新抛出 — 调试提示：只有裸 raise 才会保留异常发生位置。
• python -O 会移除 assert — 运行时验证必须使用显式的 if + raise，绝不能使用 assert。
• 异常组（3.11+）需要使用 except*，而不是 except — except ExceptionGroup 会捕获异常组，但不会拆开；except* 会选择性地处理子异常。
• 即使在 return 之后，finally 也会运行，且如果其中有自己的 return，就会覆盖返回值 — 在 finally 中意外返回会吞掉异常。