Python · 异常体系与 traceback

场景 1 · AppError(code, http_status) + API 层统一异常映射

错误码散落在几十个 handler 里各写各的, 排障与前端对接都是灾难。定义异常层级, 在最外层一次性翻译成 HTTP:

class AppError(Exception):
    code = "internal"; http_status = 500
    def __init__(self, msg, **ctx):
        super().__init__(msg)
        self.ctx = ctx               # ctx 进日志不进响应 — 防泄漏

class NotFound(AppError):  code, http_status = "not_found", 404
class Conflict(AppError): code, http_status = "conflict",  409
class RateLimited(AppError): code, http_status = "rate_limited", 429

@app.exception_handler(AppError)             # 全项目唯一的出口
async def on_app_error(req, exc):
    log.warning("app_error", code=exc.code, ctx=exc.ctx, exc_info=True)
    return JSONResponse({"code": exc.code}, status_code=exc.http_status)

场景 2 · 千万行 ETL: EAFP 风格提升脏数据吞吐

逐行预检 (get+None 判断) 每行多两次字典查找; 3.11+ 无异常时 try 零开销, 脏行占比低时先斩后奏更快:

total = 0
skipped = 0
for row in rows:                        # 千万行, 99.9% 是正常整数
    try:
        total += int(row["qty"])          # 直接上手, 不预检
    except (KeyError, ValueError):       # 只精确捕这两类脏数据
        skipped += 1
        continue                       # 单行粒度隔离, 一条脏不废一批
metrics.gauge("etl.skipped", skipped)

对比预检版: 全量耗时 -18%; 若脏行占比超过几成, 反转用 get 预检 — 异常只留给真正"异常"的路径。

场景 3 · TaskGroup 并发采集: 一组异常聚合全部失败原因

并发抓 200 个源, 旧写法 gather(return_exceptions=True) 拿一坨列表自己分拣; TaskGroup + except* 按类型各取所需:

import asyncio

async def fetch_all(urls):
    async with asyncio.TaskGroup() as tg:    # 等全部结束, 失败不丢
        tasks = [tg.create_task(fetch(u)) for u in urls]
    return [t.result() for t in tasks]

try:
    asyncio.run(fetch_all(urls))
except* TimeoutError as eg:                  # 只拿超时子组, 互不影响
    log.warning(f"{len(eg.exceptions)} sources timed out")
except* aiohttp.ClientError as eg:
    metrics.count("fetch.client_error", len(eg.exceptions))

场景 4 · 结构化错误日志: exc_info + Sentry 上报

支付失败只记一句 str(e) 等于没记。字段进 extra 供 ELK 检索, 栈走 exc_info, Sentry 按指纹聚合同类错误:

try:
    charge(order)
except PaymentError as e:
    log.error("payment_failed",
        extra={"order_id": order.id, "psp": e.psp,     # 结构化字段可检索
               "retryable": e.retryable},
        exc_info=True)                    # 全栈必须留 — str(e) 什么都没剩下
    capture_exception(e)                      # Sentry: 同类错误聚成一个 issue
    raise                                   # 原样续抛, 栈不动

场景 5 · raise ... from e: 解析层包装不丢根因

把 JSONDecodeError 包成业务异常时最常见的失误是丢链 — 排障时只剩最后一句话, 原始位置没了:

def load_config(raw: bytes):
    try:
        return json.loads(raw)
    except ValueError as e:                  # JSONDecodeError 是其子类
        raise ConfigError("config invalid") from e   # __cause__ 保留解析栈

# 日志里看到完整因果:
# ConfigError: config invalid
# The above exception was the direct cause of...
# json.decoder.JSONDecodeError: Expecting ',' delimiter: line 3 column 40

场景 6 · 事务/锁的 finally 清理模板

commit/rollback/close 三件事各有归属: 异常时回滚、无论如何关连接, 用裸 raise 保证栈不被改写:

def run_in_tx(fn):
    tx = db.begin()
    try:
        result = fn(tx)
        tx.commit()
        return result                   # 正常返回 (也要等 finally 跑完)
    except Exception:
        tx.rollback()
        raise                               # 裸 raise: 栈起点不动, 继续上抛
    finally:
        tx.close()                            # commit/rollback/异常三条路都关连接

场景 7 · 批量同步的"部分失败"策略

一万条记录同步, 不能一条坏废一批, 也不能静默跳过。收集失败项继续跑, 结尾统一报告:

failed = []
for rec in records:
    try:
        sync(rec)                             # 粒度 = 单条, 隔离失败半径
    except ValidationError as e:
        failed.append({"id": rec.id, "err": str(e)[:200]})
        metrics.incr("sync.rejected")      # 实时指标, 不等结尾
if failed:
    notify_ops(f"sync: {len(failed)}/{len(records)} failed", failed[:50])
    # 失败率超阈值再中断整批 — 策略显式写在代码里

场景 8 · tenacity 白名单重试: 只重可重试异常

裸 retry 见异常就重, 401/参数错误也重五次, 雪崩放大配额打爆。重试必须限定异常类型:

from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential

@retry(retry=retry_if_exception_type((ConnectionError, TimeoutError)),  # 白名单
       stop=stop_after_attempt(4),
       wait=wait_exponential(multiplier=0.3),            # 0.3/0.6/1.2s 退避
       reraise=True)                          # 用尽后抛原始异常, 不包 RetryError
def fetch(url): ...

# 4xx/验证失败/解析错误绝不重试 — 那不是"暂时性"故障

场景 9 · 第三方 SDK 异常翻译层

业务代码里到处 import boto3 的异常类型, 换存储引擎就要改全站。在边界处一次性翻译成领域异常:

class Storage:                                 # SDK 异常不许穿透到业务层
    def get(self, key):
        try:
            return self.s3.get_object(Bucket=self.b, Key=key)["Body"].read()
        except self.s3.exceptions.NoSuchKey:
            raise NotFound(key) from None    # 已知缺键: 领域异常, 掐断噪音链
        except EndpointConnectionError as e:
            raise StorageUnavailable(str(e)) from e  # 未知: 包一层带上 下文

场景 10 · 生产兜底中间件: unhandled → 500 + request_id 关联

总有漏网的异常。最外层中间件统一接住: 全栈进日志、响应只回安全信息、request_id 把一次请求的所有日志串起来:

@app.middleware("http")
async def catch_all(req, call_next):
    rid = req.headers.get("x-request-id", new_rid())     # 贯穿全链路的关联键
    try:
        resp = await call_next(req)
    except Exception:                              # 兜底 — 但必须留全栈+告警
        log.exception("unhandled", request_id=rid)     # exc_info: 现场一个不丢
        alert.fire("unhandled_500", rid=rid)
        return JSONResponse({"code": "internal", "request_id": rid},
                             status_code=500)        # 不回栈给客户端, 防泄漏
    resp.headers["x-request-id"] = rid
    return resp