# exceptions.py # 业务异常体系 + 全局异常处理器。 # # 原版的问题：每个 route 各自 try/except，DAO 抛什么、API 捕什么全靠人记， # 漏捕的地方直接把 SQLAlchemy 堆栈 500 给前端，既难看又泄露表结构。 # # 改造后：DAO 只负责抛「业务异常」，main.py 注册一次处理器，统一翻译成 HTTP 响应。 import logging from fastapi import FastAPI, Request, status from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from sqlalchemy.exc import DataError, IntegrityError, OperationalError, SQLAlchemyError logger = logging.getLogger(__name__) # ============================ 业务异常 ============================ class AppException(Exception): """所有业务异常的基类。 code 业务错误码，前端按码判断，不依赖文案 message 给用户看的中文提示 http_status 对应 HTTP 状态码 """ code: int = 40000 http_status: int = status.HTTP_400_BAD_REQUEST message: str = "请求处理失败" def __init__(self, message: str | None = None, *, code: int | None = None): if message: self.message = message if code is not None: self.code = code super().__init__(self.message) class NotFoundError(AppException): code = 40400 http_status = status.HTTP_404_NOT_FOUND message = "资源不存在" class ConflictError(AppException): """唯一键冲突：学号重复、班级重名、重复就业等。""" code = 40900 http_status = status.HTTP_409_CONFLICT message = "数据冲突" class DuplicateStudentNo(ConflictError): code = 40901 message = "学号已存在" class BusinessRuleError(AppException): """业务规则不满足：如"已删除的学生不允许修改""班级下还有学生不允许删除"。""" code = 42200 # 直接写 422，不用 status.HTTP_422_UNPROCESSABLE_ENTITY —— # Starlette 已经把它标记为 deprecated 并改名，跟着写会一直吃 DeprecationWarning http_status = 422 message = "业务规则校验失败" class DataAccessError(AppException): code = 50001 http_status = status.HTTP_500_INTERNAL_SERVER_ERROR message = "数据访问失败" # 兼容旧代码：原 dao/student_dao.py 里定义过 BusinessException BusinessException = AppException # ============================ 统一响应体 ============================ def ok(data=None, message: str = "success", code: int = 0) -> JSONResponse: return JSONResponse(status_code=200, content={"code": code, "message": message, "data": data}) def fail(code: int, message: str, http_status: int) -> JSONResponse: return JSONResponse(status_code=http_status, content={"code": code, "message": message, "data": None}) # ============================ 数据库异常翻译 ============================ # MySQL 的错误码含义（这些是"业务能看懂的错误"，不该返回 500）： MYSQL_ERROR_MAP = { 1062: (40900, "数据已存在（唯一键冲突）", 409), # ER_DUP_ENTRY 1451: (40900, "该记录被其他数据引用，无法删除", 409), # ER_ROW_IS_REFERENCED_2 1452: (42200, "关联的数据不存在（外键校验失败）", 422), # ER_NO_REFERENCED_ROW_2 3819: (42201, "数据不满足字段约束，请检查手机号等格式要求", 422), # ER_CHECK_CONSTRAINT_VIOLATED 1265: (40002, "字段值不合法：类型、长度或枚举取值不正确", 400), # ER_WARN_DATA_TRUNCATED 1406: (40003, "字段内容超出长度限制", 400), # ER_DATA_TOO_LONG 1048: (40004, "必填字段不能为空", 400), # ER_BAD_NULL_ERROR } def classify_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, str, int]: """把 SQLAlchemy 异常翻译成 (业务错误码, 中文提示, HTTP 状态码)。 单独抽成纯函数是为了能直接被测试覆盖 —— 真实 MySQL 的 CHECK 约束违反 抛的是 OperationalError(3819) 而不是 IntegrityError，这个坑不写测试很容易再踩。 """ orig = getattr(exc, "orig", None) err_code = None if orig is not None: args = getattr(orig, "args", None) or () if args and isinstance(args[0], int): err_code = args[0] if err_code in MYSQL_ERROR_MAP: return MYSQL_ERROR_MAP[err_code] # SQLite 没有数字错误码，按异常类型区分 if isinstance(exc, IntegrityError): return 40900, "数据违反唯一性或外键约束，请检查关联数据", 409 if isinstance(exc, DataError): return 40002, "字段值不合法：类型、长度或枚举取值不正确", 400 # 剩下的多半是连接断了、SQL 语法错误、表不存在 —— 这才是真正的 500 return 50001, "数据库暂时不可用，请稍后重试", 500 # ============================ 注册处理器 ============================ def register_exception_handlers(app: FastAPI) -> None: @app.exception_handler(AppException) async def _app_exception_handler(request: Request, exc: AppException): logger.warning("业务异常 %s %s -> [%s] %s", request.method, request.url.path, exc.code, exc.message) return fail(exc.code, exc.message, exc.http_status) @app.exception_handler(RequestValidationError) async def _validation_handler(request: Request, exc: RequestValidationError): # 把 Pydantic 的错误压成一行给人看的中文提示 first = exc.errors()[0] if exc.errors() else {} loc = ".".join(str(x) for x in first.get("loc", []) if x not in ("body", "query", "path")) msg = first.get("msg", "参数不合法") detail = f"参数校验失败：{loc} {msg}" if loc else f"参数校验失败：{msg}" logger.info("参数校验失败 %s %s -> %s", request.method, request.url.path, detail) return fail(40001, detail, status.HTTP_400_BAD_REQUEST) @app.exception_handler(IntegrityError) async def _integrity_handler(request: Request, exc: IntegrityError): # 唯一键 / 外键冲突：MySQL 报 1062 / 1451 / 1452，统一翻成 409 code, message, http_status = classify_sqlalchemy_error(exc) logger.warning("数据库约束冲突 %s %s -> %s", request.method, request.url.path, exc.orig) return fail(code, message, http_status) @app.exception_handler(DataError) async def _data_error_handler(request: Request, exc: DataError): # 字段值不合法：MySQL 1265(枚举取值非法/数据被截断)、1406(超长) 都走这里。 # 正常情况这些会在 Pydantic 层就被拦掉，这里是最后一道防线。 code, message, http_status = classify_sqlalchemy_error(exc) logger.warning("字段值非法 %s %s -> %s", request.method, request.url.path, exc.orig) return fail(code, message, http_status) @app.exception_handler(OperationalError) async def _operational_handler(request: Request, exc: OperationalError): """注意：MySQL 的 CHECK 约束违反(3819) 抛的是 OperationalError，不是 IntegrityError。 这是只有拿真库跑才会发现的事 —— 用 SQLite 测永远测不出来。 如果不单独处理，用户把手机号填成 "123" 会收到 500「服务器内部错误」， 而真正的原因只是一个字段格式问题。 """ code, message, http_status = classify_sqlalchemy_error(exc) if http_status < 500: logger.warning("字段约束违反 %s %s -> %s", request.method, request.url.path, exc.orig) else: logger.exception("数据库连接或执行异常") return fail(code, message, http_status) @app.exception_handler(SQLAlchemyError) async def _sqlalchemy_handler(request: Request, exc: SQLAlchemyError): code, message, http_status = classify_sqlalchemy_error(exc) logger.exception("数据库异常") return fail(code, message, http_status) @app.exception_handler(Exception) async def _fallback_handler(request: Request, exc: Exception): logger.exception("未捕获异常") return fail(50000, "服务器内部错误", status.HTTP_500_INTERNAL_SERVER_ERROR)