异常处理规范

异常分级、捕获、传播、全局异常处理

异常分级

按影响范围分级

[Standard] 所有自定义异常必须继承自项目统一基类，并按业务域划分异常层级。

Exception（Python 内置）
└── AppError（项目基类，所有自定义异常的父类）
    ├── BusinessError（业务异常，可预期，需返回用户）
    │   ├── ValidationError（参数校验失败）
    │   ├── NotFoundError（资源不存在）
    │   └── PermissionDeniedError（权限不足）
    └── SystemError（系统异常，非预期，需告警）
        ├── DatabaseError
        ├── RemoteCallError
        └── TimeoutError

按可恢复性分级

级别	说明	处理方式	HTTP 状态码
业务异常	用户操作不当导致，可恢复	提示用户，不告警	4xx
系统异常	依赖服务故障等，可重试	重试 / 降级	503
未知异常	代码 bug，不可预期	记录堆栈，告警	500

[Requirement] 自定义异常类必须以 Error 结尾，且包含错误码（code）和用户可读消息（message）两个字段。

示例：

class AppError(Exception):
    code: str = "INTERNAL_ERROR"
    message: str = "系统内部错误"
 
    def __init__(self, message: str | None = None):
        if message:
            self.message = message
        super().__init__(self.message)
 
class BusinessError(AppError):
    code = "BUSINESS_ERROR"
 
class NotFoundError(BusinessError):
    code = "NOT_FOUND"
    message = "资源不存在"

捕获原则

捕获范围

[Requirement] 禁止裸 except:（捕获所有异常包括 KeyboardInterrupt），必须指定具体异常类型。

[Caution] 禁止捕获 Exception 后直接吞掉（pass），至少要记录日志。

[Standard] 只捕获你能处理的异常，不能处理的让它向上传播。

示例：

# 正确：只捕获能处理的异常
def load_config(path: str) -> dict:
    try:
        with open(path) as f:
            return yaml.safe_load(f)
    except FileNotFoundError:
        logger.warning("配置文件不存在，使用默认配置: %s", path)
        return {}
    except yaml.YAMLError as e:
        logger.error("配置文件解析失败: %s", path, exc_info=True)
        raise AppError("配置文件格式错误") from e

异常信息

[Requirement] 捕获异常后必须记录日志，日志中包含异常堆栈（exc_info=True 或 logger.exception）。

[Standard] 重新抛出异常时使用 raise ... from e 保留原始异常链。

反模式：

# 错误：裸捕获并吞掉
try:
    risky_call()
except:
    pass
 
# 错误：捕获 Exception 后继续
try:
    risky_call()
except Exception:
    print("出错了")

传播规则

[Standard] 底层异常不应直接暴露给上层调用方，应在边界层（如 Service 层）转换为业务异常。

[Requirement] 跨进程/跨服务调用时，必须将底层异常（如 HTTP 错误、数据库错误）转换为领域异常，避免泄漏实现细节。

示例：

# 正确：边界层转换异常
class UserService:
    def get_user(self, user_id: int) -> User:
        try:
            return self.repo.get(user_id)
        except sqlalchemy.exc.NoResultFound:
            raise NotFoundError(f"用户 {user_id} 不存在")
        except sqlalchemy.exc.DBAPIError as e:
            logger.error("数据库查询失败", exc_info=True)
            raise SystemError("数据库异常") from e

重试场景

[Default] 对网络调用、消息消费等可重试场景，使用指数退避策略，最多重试 3 次。

[Caution] 只有幂等操作才能重试；非幂等操作（如扣款、下单）禁止自动重试。

import backoff
 
@backoff.on_exception(backoff.expo, requests.RequestException, max_tries=3)
def call_remote_api(url: str) -> dict:
    resp = requests.get(url, timeout=5)
    resp.raise_for_status()
    return resp.json()

全局异常处理

[Requirement] Web 应用必须配置全局异常处理器，统一捕获未处理的异常，返回标准错误响应。

[Standard] 全局异常处理器不应吞掉未知异常时，必须记录完整堆栈并触发告警。

标准错误响应格式：

{
  "code": "USER_NOT_FOUND",
  "message": "用户不存在",
  "request_id": "req-abc123",
  "timestamp": "2024-01-01T00:00:00Z"
}

示例（FastAPI）：

@app.exception_handler(AppError)
async def app_error_handler(request: AppError):
    logger.exception("业务异常: code=%s", exc.code)
    return JSONResponse(
        status_code=exc.http_status,
        content={"code": exc.code, "message": exc.message,
                 "request_id": request.state.request_id},
    )
 
@app.exception_handler(Exception)
async def unhandled_error_handler(request, exc):
    logger.exception("未处理异常", exc_info=True)
    alert.send(exc)  # 触发告警
    return JSONResponse(
        status_code=500,
        content={"code": "INTERNAL_ERROR", "message": "系统内部错误",
                 "request_id": request.state.request_id},
    )

反模式

反模式	示例	问题
裸 `except`	`except: pass`	吞掉所有异常，包括 KeyboardInterrupt，难以调试
捕获 Exception 后吞掉	`except Exception: pass`	隐藏错误，问题无法发现
仅打印不抛出	`except Exception as e: print(e)`	没有堆栈，丢失上下文
异常当控制流	`try: ... except ValueError: ...` 用于正常分支判断	异常机制效率低且语义不清，应用 if 判断
过度宽泛的捕获	`except Exception` 然后什么都做	应该只捕获能处理的类型
丢失原始异常	`raise NewError("msg")` 不加 `from`	异常链断裂，根因丢失
返回错误码代替异常	`return None` 表示失败	调用方容易忘记检查，应抛异常
在循环里捕获 Exception	`for x in ...: try: ... except: pass`	静默失败，数据不一致