强曰为道

与天地相似,故不违。知周乎万物,而道济天下,故不过。旁行而不流,乐天知命,故不忧.
文档目录

10 - 错误处理

第 10 章:错误处理

掌握 Python 的异常处理机制,编写健壮、可维护的代码。


10.1 异常基础

10.1.1 常见异常类型

异常说明示例
ValueError值不正确int("abc")
TypeError类型不匹配"1" + 2
KeyError字典键不存在{}["key"]
IndexError索引越界[][0]
AttributeError属性不存在None.x
FileNotFoundError文件不存在open("不存在.txt")
ImportError导入失败import 不存在的模块
ZeroDivisionError除以零1 / 0
RuntimeError运行时错误一般性运行错误
StopIteration迭代器耗尽next(iter([]))

10.1.2 异常层次

BaseException
├── SystemExit
├── KeyboardInterrupt
├── GeneratorExit
└── Exception
    ├── ValueError
    ├── TypeError
    ├── KeyError
    ├── IndexError
    ├── AttributeError
    ├── OSError
    │   ├── FileNotFoundError
    │   ├── PermissionError
    │   └── FileExistsError
    ├── RuntimeError
    ├── StopIteration
    ├── ArithmeticError
    │   ├── ZeroDivisionError
    │   └── OverflowError
    └── ...

10.2 try-except 语句

10.2.1 基本语法

try:
    result = 10 / int(input("输入一个数字: "))
    print(f"结果: {result}")
except ValueError:
    print("请输入有效数字!")
except ZeroDivisionError:
    print("不能除以零!")

10.2.2 捕获多个异常

# 方式一:分别处理
try:
    data = json.loads(user_input)
except json.JSONDecodeError:
    print("JSON 格式错误")
except ValueError:
    print("值错误")

# 方式二:合并处理
try:
    process(data)
except (ValueError, TypeError) as e:
    print(f"数据错误: {e}")

10.2.3 获取异常信息

try:
    result = 10 / 0
except ZeroDivisionError as e:
    print(f"异常类型: {type(e).__name__}")  # ZeroDivisionError
    print(f"异常消息: {e}")                  # division by zero
    print(f"异常参数: {e.args}")             # ('division by zero',)

10.2.4 完整的 try 语句

try:
    result = risky_operation()
except ValueError as e:
    handle_value_error(e)
except (TypeError, RuntimeError) as e:
    handle_other_error(e)
except Exception as e:
    handle_unexpected_error(e)
else:
    # 没有异常时执行
    print(f"成功: {result}")
finally:
    # 无论如何都执行(清理资源)
    cleanup()

10.2.5 异常链

# 原始异常被保留(Python 3)
try:
    try:
        result = int("abc")
    except ValueError as original:
        raise RuntimeError("数据转换失败") from original
except RuntimeError as e:
    print(f"错误: {e}")
    print(f"原因: {e.__cause__}")

# 抑制异常链
try:
    result = int("abc")
except ValueError:
    raise RuntimeError("数据转换失败") from None  # 不显示原始异常

10.3 自定义异常

10.3.1 定义异常层次

class AppError(Exception):
    """应用基础异常。"""
    pass

class ValidationError(AppError):
    """数据验证异常。"""
    def __init__(self, field: str, message: str):
        self.field = field
        self.message = message
        super().__init__(f"验证失败 - {field}: {message}")

class NotFoundError(AppError):
    """资源不存在异常。"""
    def __init__(self, resource: str, identifier: str):
        self.resource = resource
        self.identifier = identifier
        super().__init__(f"{resource} '{identifier}' 不存在")

class PermissionError(AppError):
    """权限不足异常。"""
    pass

# 使用
def get_user(user_id: int) -> dict:
    if user_id <= 0:
        raise ValidationError("user_id", "必须为正整数")
    
    users = {1: {"name": "Alice"}, 2: {"name": "Bob"}}
    
    if user_id not in users:
        raise NotFoundError("用户", str(user_id))
    
    return users[user_id]

try:
    user = get_user(99)
except NotFoundError as e:
    print(e)        # 用户 '99' 不存在
    print(e.resource)   # 用户
    print(e.identifier) # 99
except ValidationError as e:
    print(e)        # 验证失败 - user_id: 必须为正整数

10.3.2 带错误码的异常

from enum import IntEnum

class ErrorCode(IntEnum):
    UNKNOWN = 1000
    VALIDATION_ERROR = 1001
    NOT_FOUND = 1002
    PERMISSION_DENIED = 1003
    RATE_LIMITED = 1004

class APIError(Exception):
    def __init__(self, code: ErrorCode, message: str, details: dict | None = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)
    
    def to_dict(self) -> dict:
        return {
            "error": {
                "code": self.code.value,
                "message": self.message,
                "details": self.details,
            }
        }

# 使用
raise APIError(
    ErrorCode.VALIDATION_ERROR,
    "请求参数无效",
    {"field": "email", "reason": "格式不正确"}
)

10.4 上下文管理器

10.4.1 with 语句

# 文件操作
with open("data.txt") as f:
    content = f.read()
# 文件自动关闭

# 多个上下文管理器
with open("input.txt") as fin, open("output.txt", "w") as fout:
    fout.write(fin.read())

10.4.2 自定义上下文管理器

class DatabaseConnection:
    """数据库连接上下文管理器。"""
    
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None
    
    def __enter__(self):
        print(f"连接数据库: {self.connection_string}")
        self.connection = {"connected": True, "string": self.connection_string}
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"断开数据库连接")
        self.connection = None
        # 返回 True 会抑制异常,False/None 不抑制
        return False

with DatabaseConnection("postgresql://localhost/mydb") as conn:
    print(f"查询中... 连接状态: {conn['connected']}")
# 断开数据库连接(即使发生异常也会执行)

10.4.3 contextlib 模块

from contextlib import contextmanager, suppress, redirect_stdout
import io

# @contextmanager 装饰器
@contextmanager
def timer(label: str):
    import time
    start = time.perf_counter()
    try:
        yield  # yield 的值会赋给 as 变量
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.4f} 秒")

with timer("数据处理"):
    time.sleep(0.5)

# suppress:忽略特定异常
with suppress(FileNotFoundError):
    open("不存在的文件.txt")

# 重定向输出
buffer = io.StringIO()
with redirect_stdout(buffer):
    print("这会写入 buffer")
output = buffer.getvalue()

10.5 断言(assert)

# 语法:assert condition, message
def calculate_discount(price: float, discount: float) -> float:
    assert 0 <= discount <= 1, f"折扣必须在 0-1 之间,收到 {discount}"
    assert price >= 0, f"价格不能为负,收到 {price}"
    return price * (1 - discount)

print(calculate_discount(100, 0.2))  # 80.0
# calculate_discount(100, 1.5)  # AssertionError: 折扣必须在 0-1 之间

# ⚠️ assert 可能被 -O 优化标志禁用
# 生产环境使用显式检查
用途推荐方式
开发时检查不变量assert
验证用户输入if + raise
检查函数前置条件if + raise
检查后置条件assertif + raise

10.6 日志

10.6.1 基本使用

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)

logger = logging.getLogger(__name__)

# 使用
logger.debug("调试信息")
logger.info("一般信息")
logger.warning("警告")
logger.error("错误")
logger.critical("严重错误")

# 带异常信息
try:
    result = 1 / 0
except ZeroDivisionError:
    logger.exception("计算错误")  # 自动记录堆栈信息

10.6.2 日志级别

级别数值使用场景
DEBUG10调试细节
INFO20一般操作记录
WARNING30警告(可恢复)
ERROR40错误(功能受损)
CRITICAL50严重错误(系统崩溃)

10.6.3 配置日志处理器

import logging
from logging.handlers import RotatingFileHandler

# 创建 logger
logger = logging.getLogger("myapp")
logger.setLevel(logging.DEBUG)

# 控制台处理器
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))

# 文件处理器(轮转)
file_handler = RotatingFileHandler(
    "app.log",
    maxBytes=10 * 1024 * 1024,  # 10MB
    backupCount=5,
    encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
))

logger.addHandler(console)
logger.addHandler(file_handler)

10.7 最佳实践

10.7.1 异常处理原则

# ✅ 精确捕获
try:
    value = my_dict[key]
except KeyError:
    value = default

# ❌ 过于宽泛
try:
    value = my_dict[key]
except Exception:
    value = default

# ✅ 保持 try 块小
# ❌ 把整个函数放在 try 里
try:
    data = parse(input)
    result = process(data)
    save(result)
except ...:
    ...

10.7.2 EAFP 原则

# EAFP: Easier to Ask Forgiveness than Permission(先做再说)

# LBYL: Look Before You Leap(三思后行)
if key in my_dict:
    value = my_dict[key]
else:
    value = default

# EAFP(Pythonic)
try:
    value = my_dict[key]
except KeyError:
    value = default

# 更好:使用 get()
value = my_dict.get(key, default)

10.7.3 常见模式

# 重试装饰器
import time
from functools import wraps

def retry(max_attempts: int = 3, delay: float = 1.0, backoff: float = 2.0):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_delay = delay
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    logger.warning(f"尝试 {attempt} 失败: {e}")
                    time.sleep(current_delay)
                    current_delay *= backoff
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5)
def fetch_data(url: str) -> dict:
    import requests
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

10.8 注意事项

🔴 注意

  • 不要捕获 BaseException 或裸 except:(会吞掉 KeyboardInterrupt
  • finally 中不要使用 return(会覆盖 try/except 的返回值)
  • assert 会被 -O 优化模式禁用,不要用于验证用户输入
  • 日志不要记录敏感信息(密码、token 等)

💡 提示

  • 异常也是 API 的一部分,文档中要说明可能抛出的异常
  • 使用 logging 而非 print 记录日志
  • 自定义异常继承 Exception,不要继承 BaseException
  • 使用 contextlib.suppress 替代空的 try/except

📌 业务场景

import logging
from contextlib import contextmanager
from dataclasses import dataclass

logger = logging.getLogger(__name__)

@dataclass
class OrderResult:
    order_id: str
    status: str
    message: str

class OrderError(Exception):
    pass

class InsufficientStockError(OrderError):
    def __init__(self, product: str, requested: int, available: int):
        self.product = product
        self.requested = requested
        self.available = available
        super().__init__(f"{product} 库存不足: 需要 {requested},可用 {available}")

@contextmanager
def order_transaction(order_id: str):
    """订单事务上下文管理器。"""
    logger.info(f"开始订单事务: {order_id}")
    try:
        yield
        logger.info(f"订单事务完成: {order_id}")
    except OrderError as e:
        logger.error(f"订单失败: {order_id} - {e}")
        raise
    except Exception as e:
        logger.critical(f"订单异常: {order_id} - {e}", exc_info=True)
        raise OrderError(f"订单处理异常: {e}") from e

def process_order(order_id: str, product: str, quantity: int, stock: int) -> OrderResult:
    with order_transaction(order_id):
        if quantity <= 0:
            raise ValueError("数量必须为正数")
        
        if quantity > stock:
            raise InsufficientStockError(product, quantity, stock)
        
        # 模拟订单处理
        logger.info(f"扣减库存: {product} x{quantity}")
        
        return OrderResult(
            order_id=order_id,
            status="success",
            message=f"订单 {order_id} 创建成功",
        )

10.9 扩展阅读