27 April 2025
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
class UnauthorizedException(HTTPException):
def __init__(self, detail: str = "未授权的访问"):
super().__init__(status_code=401, detail=detail)
async def unauthorized_exception_handler(request: Request, exc: UnauthorizedException):
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.status_code, "message": exc.detail}
)
# 添加异常处理器
app.add_exception_handler(UnauthorizedException, unauthorized_exception_handler)
class AuthMiddleware(BaseHTTPMiddleware):
def __init__(self, app):
super().__init__(app)
async def dispatch(self, request: Request, call_next):
# 检查认证(这里可以换成你的逻辑)
token = request.headers.get("Authorization")
if not token:
# 这里改成直接返回Response,避免raise
return JSONResponse(
status_code=401,
content={"code": 401, "message": "缺少认证令牌"}
)
response = await call_next(request)
return response
# 添加中间件
app.add_middleware(AuthMiddleware)
@app.get("/protected")
async def protected_route():
return {"message": "这是一个受保护的路由"}