Python中全局异常处理的必要性及实现方法

@app.get("/users/{user_id}")
def get_user(user_id: int):
    try:
        user = db.query(user_id)
        if not user:
            return {"code": 404, "msg": "用户不存在"}
        return {"code": 200, "data": user}
    except ValueError as e:
        logger.error(f"参数错误: {e}")
        return {"code": 400, "msg": str(e)}
    except Exception as e:
        logger.exception("未知错误")
        return {"code": 500, "msg": "服务器内部错误"}

# exceptions.py
class AppException(Exception):
    """应用级基础异常"""
    def __init__(self, code: int, message: str, status_code: int = 200):
        self.code = code
        self.message = message
        self.status_code = status_code


class NotFoundError(AppException):
    def __init__(self, resource: str = "资源"):
        super().__init__(code=404, message=f"{resource}不存在", status_code=404)


class AuthenticationError(AppException):
    def __init__(self, message: str = "认证失败"):
        super().__init__(code=401, message=message, status_code=401)


class ValidationError(AppException):
    def __init__(self, message: str = "参数校验失败"):
        super().__init__(code=422, message=message, status_code=422)

# main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from exceptions import AppException

app = FastAPI()

# 1. 处理所有自定义业务异常
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.code, "message": exc.message}
    )

# 2. 兜底：处理所有未预期的系统异常
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # ⚠️ 记录完整堆栈，但绝不暴露给客户端
    logger.exception(f"Unhandled exception on {request.method} {request.url.path}")
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误"}
    )

from flask import Flask, jsonify
from exceptions import AppException

app = Flask(__name__)

@app.errorhandler(AppException)
def handle_app_exception(exc):
    return jsonify(code=exc.code, message=exc.message), exc.status_code

@app.errorhandler(Exception)
def handle_global_exception(exc):
    app.logger.exception("Unhandled exception")
    return jsonify(code=500, message="服务器内部错误"), 500

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = db.query(user_id)
    if not user:
        raise NotFoundError("用户")      # ← 只管抛，不管接
    return {"code": 200, "data": user}   # ← 只写正常流程

{
  "code": 404,
  "message": "用户不存在",
  "request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}

import logging
import uuid

logger = logging.getLogger("app")

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    request_id = getattr(request.state, "request_id", str(uuid.uuid4()))
    logger.exception(
        "Unhandled exception",
        extra={
            "request_id": request_id,
            "method": request.method,
            "path": request.url.path,
            "client_ip": request.client.host,
        }
    )
    # 可选：接入 Sentry / Prometheus 告警
    # sentry_sdk.capture_exception(exc)
    
    return JSONResponse(
        status_code=500,
        content={
            "code": 500,
            "message": "服务器内部错误",
            "request_id": request_id
        }
    )