07 异常、上下文与日志

大约 3 分钟编程语言Python异常logging可观测性

07 异常、上下文与日志

异常负责改变控制流，日志负责留下证据。不要用异常打印代替错误处理，也不要用大量 except Exception 隐藏故障。

1. 捕获具体异常

from pathlib import Path

def read_text(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except FileNotFoundError:
        return ""
    except PermissionError as exc:
        raise RuntimeError(f"cannot read {path}") from exc

按可恢复方式捕获异常：

参数错误在边界处转为 ValueError 或领域异常。
临时网络故障可按策略重试。
权限、配置和程序错误通常应失败并暴露证据。
KeyboardInterrupt、SystemExit 不应被普通业务逻辑吞掉。

2. 异常链

def parse_port(raw: str) -> int:
    try:
        port = int(raw)
    except ValueError as exc:
        raise ValueError(f"invalid port: {raw!r}") from exc

    if not 1 <= port <= 65535:
        raise ValueError(f"port out of range: {port}")
    return port

raise ... from exc 保留原始原因。排查底层驱动、网络库或解析错误时，异常链比单独一行错误消息更重要。

3. 自定义异常

class OpsError(Exception):
    """Base exception for this application."""

class CheckFailed(OpsError):
    def __init__(self, check_name: str, reason: str) -> None:
        super().__init__(f"{check_name}: {reason}")
        self.check_name = check_name
        self.reason = reason

自定义异常适合调用者需要区分处理的领域错误。不要为每个内部函数创建无意义异常。

4. 重试边界

import time
from collections.abc import Callable

def retry[T](operation: Callable[[], T], attempts: int = 3, base_delay: float = 0.5) -> T:
    last_error: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            return operation()
        except (TimeoutError, ConnectionError) as exc:
            last_error = exc
            if attempt == attempts:
                break
            time.sleep(base_delay * (2 ** (attempt - 1)))
    assert last_error is not None
    raise last_error

这段代码使用 Python 3.12 起支持的泛型类型参数语法。需要兼容更早版本时，改用 TypeVar。

重试必须限定异常类型、总次数、总预算和退避策略。非幂等操作、认证失败和参数错误不应盲目重试；多实例系统还应加入随机抖动，避免同步重试。

5. 资源清理

from pathlib import Path

def count_non_empty_lines(path: Path) -> int:
    count = 0
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            if line.strip():
                count += 1
    return count

优先使用 with 和上下文管理器。finally 适合没有现成上下文管理器的成对操作，如释放特殊句柄、恢复状态。

6. 日志基础

import logging

log = logging.getLogger(__name__)

def process(order_id: str, attempt: int) -> None:
    log.info("processing order", extra={"order_id": order_id, "attempt": attempt})

应用入口只配置一次日志，库模块只获取 logger，不调用 basicConfig。

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

推荐级别：

级别	使用场景
`DEBUG`	开发诊断、详细状态，不在默认生产级别长期开启
`INFO`	关键状态变化和成功结果
`WARNING`	可继续运行但需要关注的异常
`ERROR`	当前操作失败
`CRITICAL`	服务整体不可用或数据完整性受威胁

7. 结构化日志字段

日志消息保持稳定，动态值放入结构化字段：

log.warning(
    "dependency unavailable",
    extra={"dependency": "redis", "retry_after_seconds": 5},
)

常见字段包括 request_id、trace_id、service、host、operation、duration_ms、error_type 和 retry_count。

不要记录密码、令牌、Cookie、私钥、完整请求体和未脱敏的个人信息。

8. ExceptionGroup

Python 3.11 起可以使用异常组表达多个并发失败：

def run_tasks(tasks: list[Callable[[], None]]) -> None:
    errors = []
    for task in tasks:
        try:
            task()
        except Exception as exc:
            errors.append(exc)
    if errors:
        raise ExceptionGroup("multiple tasks failed", errors)

异常组适合批量任务、结构化并发和需要一次性报告多个错误的边界。

9. 常见陷阱

except Exception: pass 吞掉所有证据。
在库代码中配置全局日志。
同时打印异常和记录日志，导致重复噪声。
将异常消息作为唯一错误分类；应保留类型和错误码。
无限制重试，制造流量风暴。
日志中记录凭据或敏感载荷。