Python错误处理Skill python-error-handling

---

name: python-error-handling description: Python错误处理模式，包括输入验证、异常层次结构和部分失败处理。适用于实现验证逻辑、设计异常策略、处理批量处理失败或构建健壮的API。

Python错误处理

构建健壮的Python应用程序，包括正确的输入验证、有意义的异常和优雅的失败处理。良好的错误处理使调试更容易，系统更可靠。

何时使用此技能

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

核心概念

1. 快速失败

尽早验证输入，在昂贵操作之前。尽可能一次性报告所有验证错误。

2. 有意义的异常

使用带有上下文的适当异常类型。消息应解释什么失败了、为什么以及如何修复。

3. 部分失败

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

4. 保持上下文

链式异常以维护完整的错误追踪，便于调试。

快速开始

def fetch_page(url: str, page_size: int) -> Page:
    if not url:
        raise ValueError("'url' 是必需的")
    if not 1 <= page_size <= 100:
        raise ValueError(f"'page_size' 必须在1-100之间，得到 {page_size}")
    # 现在可以安全继续...

基础模式

模式1：早期输入验证

在处理开始之前，在API边界验证所有输入。

def process_order(
    order_id: str,
    quantity: int,
    discount_percent: float,
) -> OrderResult:
    """使用验证处理订单。"""
    # 验证必填字段
    if not order_id:
        raise ValueError("'order_id' 是必需的")

    # 验证范围
    if quantity <= 0:
        raise ValueError(f"'quantity' 必须为正数，得到 {quantity}")

    if not 0 <= discount_percent <= 100:
        raise ValueError(
            f"'discount_percent' 必须在0-100之间，得到 {discount_percent}"
        )

    # 验证通过，继续处理
    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:
    """将字符串解析为OutputFormat枚举。

    Args:
        value: 来自用户输入的格式字符串。

    Returns:
        验证后的OutputFormat枚举成员。

    Raises:
        ValueError: 如果格式不被识别。
    """
    try:
        return OutputFormat(value.lower())
    except ValueError:
        valid_formats = [f.value for f in OutputFormat]
        raise ValueError(
            f"无效格式 '{value}'。 "
            f"有效选项: {', '.join(valid_formats)}"
        )

# 在API边界使用
def export_data(data: list[dict], format_str: str) -> bytes:
    output_format = parse_output_format(format_str)  # 快速失败
    # 函数其余部分使用类型化的OutputFormat
    ...

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

使用Pydantic模型进行结构化输入验证，并自动生成错误消息。

from pydantic import BaseModel, Field, field_validator

class CreateUserInput(BaseModel):
    """用户创建的输入模型。"""

    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("无效的电子邮件格式")
        return v.lower()

    @field_validator("name")
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return v.strip().title()

# 使用
try:
    user_input = CreateUserInput(
        email="user@example.com",
        name="john doe",
        age=25,
    )
except ValidationError as e:
    # Pydantic提供详细的错误信息
    print(e.errors())

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

适当使用Python内置异常类型，根据需要添加上下文。

失败类型	异常	示例
无效输入	ValueError	错误的参数值
错误类型	TypeError	预期字符串，得到整数
缺失项	KeyError	字典键未找到
操作失败	RuntimeError	服务不可用
超时	TimeoutError	操作耗时过长
文件未找到	FileNotFoundError	路径不存在
权限被拒绝	PermissionError	访问被禁止

# 好：带上下文的特定异常
raise ValueError(f"'page_size' 必须在1-100之间，得到 {page_size}")

# 避免：通用异常，无上下文
raise Exception("无效参数")

高级模式

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

创建携带结构化信息的领域特定异常。

class ApiError(Exception):
    """API错误的基础异常。"""

    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):
    """当超出速率限制时引发。"""

    def __init__(self, retry_after: int) -> None:
        self.retry_after = retry_after
        super().__init__(
            f"超出速率限制。请在 {retry_after}秒后重试",
            status_code=429,
        )

# 使用
def handle_response(response: Response) -> dict:
    match response.status_code:
        case 200:
            return response.json()
        case 401:
            raise ApiError("无效凭据", 401)
        case 404:
            raise ApiError(f"资源未找到: {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"客户端错误: {response.text}", code)
        case code if code >= 500:
            raise ApiError(f"服务器错误: {response.text}", code)

模式6：异常链

重新引发时保留原始异常，以维护调试跟踪。

import httpx

class ServiceError(Exception):
    """高级服务操作失败。"""
    pass

def upload_file(path: str) -> str:
    """上传文件并返回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"上传失败: 在 '{path}' 未找到文件") from e
    except httpx.HTTPStatusError as e:
        raise ServiceError(
            f"上传失败: 服务器返回 {e.response.status_code}"
        ) from e
    except httpx.RequestError as e:
        raise ServiceError(f"上传失败: 网络错误") from e

模式7：批量处理与部分失败

不要让一个坏项目中止整个批次。按项目跟踪结果。

from dataclasses import dataclass

@dataclass
class BatchResult[T]:
    """批量处理的结果。"""

    succeeded: dict[int, T]  # 索引 -> 结果
    failed: dict[int, Exception]  # 索引 -> 错误

    @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]:
    """处理项目，捕获个别失败。

    Args:
        items: 要处理的项目。

    Returns:
        BatchResult，包含按索引的成功和失败项目。
    """
    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)

# 调用者处理部分结果
result = process_batch(items)
if not result.all_succeeded:
    logger.warning(
        f"批量完成，有 {result.failure_count} 个失败",
        failed_indices=list(result.failed.keys()),
    )

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

在不将业务逻辑耦合到UI的情况下，提供批量进度的可见性。

from collections.abc import Callable

ProgressCallback = Callable[[int, int, str], None]  # 当前，总计，状态

def process_large_batch(
    items: list[Item],
    on_progress: ProgressCallback | None = None,
) -> BatchResult:
    """使用可选进度报告处理批量。

    Args:
        items: 要处理的项目。
        on_progress: 可选回调，接收 (当前, 总计, 状态)。
    """
    total = len(items)
    succeeded = {}
    failed = {}

    for idx, item in enumerate(items):
        if on_progress:
            on_progress(idx, total, f"处理 {item.id}")

        try:
            succeeded[idx] = process_single_item(item)
        except Exception as e:
            failed[idx] = e

    if on_progress:
        on_progress(total, total, "完成")

    return BatchResult(succeeded=succeeded, failed=failed)

最佳实践总结

1. 尽早验证 - 在昂贵操作之前检查输入
2. 使用特定异常 - ValueError, TypeError, 而不是通用 Exception
3. 包括上下文 - 消息应解释什么、为什么以及如何修复
4. 在边界转换类型 - 尽早将字符串解析为枚举/域类型
5. 链式异常 - 使用 raise ... from e 以保留调试信息
6. 处理部分失败 - 不要因单个项目错误而中止批次
7. 使用Pydantic - 用于具有结构化错误的复杂输入验证
8. 记录失败模式 - 文档字符串应列出可能的异常
9. 带上下文日志 - 包括ID、计数和其他调试信息
10. 测试错误路径 - 验证异常是否正确引发