如何在项目全局抛出自定义异常？无需逐个函数添加装饰器

要在FastAPI项目中全局层面自动抛出指定异常（无需给每个函数加装饰器），可以通过全局依赖、中间件、Pydantic校验这几种方式实现，对应不同场景：

1. 先定义自定义异常类 #

首先统一封装你的业务异常，继承FastAPI的HTTPException以便框架自动处理：

from fastapi import HTTPException

class NOT_AUTHORIZED(HTTPException):
    def __init__(self):
        super().__init__(status_code=401, detail="未授权访问")

class INVALID_SQL(HTTPException):
    def __init__(self, detail="无效的SQL参数"):
        super().__init__(status_code=400, detail=detail)

2. 未授权场景：全局依赖自动校验 #

通过全局依赖实现所有路由的统一授权校验，不符合条件时自动抛出NOT_AUTHORIZED：

from fastapi import FastAPI, Request

app = FastAPI()

# 授权校验逻辑
def validate_token(token: str) -> bool:
    # 这里替换成你的实际token验证逻辑
    return token == "valid_token"

async def global_auth_check(request: Request):
    auth_token = request.headers.get("Authorization", "").replace("Bearer ", "")
    if not auth_token or not validate_token(auth_token):
        raise NOT_AUTHORIZED()

# 将校验依赖全局绑定到所有路由
app.dependency_overrides[None] = [global_auth_check]

这样所有接口请求都会先经过global_auth_check的校验，未通过则直接抛出异常。

3. 无效SQL参数场景：两种实现方式 #

方式一：全局中间件拦截校验 #

通过中间件统一拦截所有请求，检查SQL相关参数并抛出异常：

from starlette.middleware.base import BaseHTTPMiddleware

def is_valid_sql_param(param: str) -> bool:
    # 替换成你的实际SQL参数校验逻辑，比如禁止非法关键字、检查格式
    return "DROP" not in param.upper()

class SQLParamMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 根据请求方法获取参数
        if request.method == "GET":
            params = dict(request.query_params)
        else:
            try:
                params = await request.json()
            except:
                params = {}
        
        sql_param = params.get("sql_param")
        if sql_param and not is_valid_sql_param(sql_param):
            raise INVALID_SQL(detail=f"非法SQL参数：{sql_param}")
        
        response = await call_next(request)
        return response

# 注册中间件
app.add_middleware(SQLParamMiddleware)

方式二：Pydantic模型字段校验 #

如果SQL参数是通过请求体/查询参数传入的，用Pydantic的字段校验规则自动抛出异常：

from pydantic import BaseModel, field_validator

class SQLQueryRequest(BaseModel):
    sql_param: str

    @field_validator("sql_param")
    def check_sql_param(cls, value):
        if "DROP" in value.upper():
            raise INVALID_SQL(detail=f"参数包含非法SQL关键字：{value}")
        return value

# 路由中直接使用模型，参数不合法时自动抛出异常
@app.post("/execute-sql")
async def execute_sql(request: SQLQueryRequest):
    # 业务逻辑
    return {"status": "success", "param": request.sql_param}

补充：全局统一异常响应格式（可选） #

如果需要统一所有异常的返回格式，可以注册全局异常处理器：

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": exc.detail, "data": None}
    )

内容的提问来源于stack exchange，提问作者daniel guo