在使用 FastAPI 开发接口时,我们通常很快就能完成基本的 CRUD 功能。但随着项目复杂度提升,仅仅会写接口是不够的,还需要处理一些工程化问题,比如异常统一处理、请求拦截、公共依赖管理以及接口测试。
这些能力主要依赖 FastAPI 的几个核心机制:
- 异常处理(Exception Handling)
- 中间件(Middleware)
- 依赖注入(Dependency Injection)
掌握这些内容,可以让你的 FastAPI 项目从“能用”提升到“规范、可维护”。 接下来我们逐个来看。
全局统一返回结构(Result)
在实际项目中,为了让前后端交互更规范,通常会统一接口返回格式,比如:
Result(code,msg,data){"code":200,"msg":"success","data":{}}定义统一返回对象
可以先封装一个通用结构:
fromtypingimportAny,Generic,TypeVarfrompydanticimportBaseModel T=TypeVar("T")classResult(BaseModel,Generic[T]):code:int=200msg:str="success"data:T|None=None@classmethoddefok(cls,data:T|None=None,msg:str="success",code:int=200)->"Result[T]":returncls(code=code,msg=msg,data=data)@classmethoddeferror(cls,code:int,msg:str,data:Any=None)->"Result[Any]":returncls(code=code,msg=msg,data=data)接口中使用
@app.get("/user")defget_user():returnResult.success(data={"name":"Tom"})异常处理(Exception Handling)
FastAPI 的异常处理分两层:
内置异常:HTTPException
这是最常用的:
@router.get("/detail/{user_id}")asyncdefget_user_detail(user_id:int):# 路径参数会根据类型注解自动转换ifuser_id!=1:raiseHTTPException(status_code=404,detail="User not found")return{"user_id":user_id}当我们访问user_id为2的用户就会发生异常
自定义异常 + 全局处理
你可以定义自己的异常:
classMyException(Exception):def__init__(self,msg:str):self.msg=msg然后注册全局异常处理器:
fromfastapiimportRequestfromfastapi.responsesimportJSONResponse# app = FastAPI()@app.exception_handler(MyException)asyncdefmy_exception_handler(request:Request,exc:MyException):returnJSONResponse(content=Result.error(code=500,msg=exc.msg).model_dump())注意点:
@app.exception_handler(...)是全局生效- 只要是这个 app 下的所有路由都会生效
修改接口
@router.get("/detail/{user_id}")asyncdefget_user_detail(user_id:int):ifuser_id!=1:raiseMyException("用户id异常")return{"user_id":user_id}进行测试
Pydantic 校验异常也统一
pydantic校验不通过会抛出RequestValidationError,我们对RequestValidationError也进行处理。
注册异常处理器
@app.exception_handler(RequestValidationError)asyncdefvalidation_exception_handler(request:Request,exc:RequestValidationError):returnJSONResponse(content=Result.error(code=422,msg="参数校验失败",data=exc.errors()).model_dump())用之前的接口测试
classUserCreateRequest(BaseModel):# 定义请求体字段包含参数校验user_name:str=Field(min_length=2,max_length=20)age:int=Field(ge=0,le=150)@router.post("")asyncdefcreate_user(request:UserCreateRequest):returnrequest.model_dump()测试一下返回结果
中间件(Middleware)
中间件本质是:请求进来 → 处理 → 响应出去 的“拦截器”
执行顺序
请求 → middleware1 → middleware2 → 路由 → middleware2 → middleware1 → 返回能干嘛
- 记录日志(log)
- 鉴权(token校验)
- 请求耗时统计
- CORS(跨域)
- Trace ID(链路追踪)
最基础中间件
fromfastapiimportRequest# "http" 不是名字,而是“中间件类型@app.middleware("http")asyncdeflog_middleware(request:Request,call_next):print("请求来了:",request.url)response=awaitcall_next(request)print("响应返回")returnresponse@app.middleware("http")asyncdeflog_middleware2(request:Request,call_next):print("请求来了2:",request.url)response=awaitcall_next(request)print("响应返回2")returnresponse调用一下http://localhost:8000/user/detail/2接口,并查看打印日志
请求来了2: http://localhost:8000/user/detail/2 请求来了: http://localhost:8000/user/detail/2 响应返回 响应返回2这还有一个美名叫(洋葱模型)
log_middleware2 ← 外层(最后定义) log_middleware route log_middleware log_middleware2 ← 外层返回依赖注入
在 FastAPI 中,依赖注入(Depends)就是:把函数执行结果自动注入到参数里 ,自动传给接口,而不是在接口里手动创建。
依赖注入的常见用途
获取数据库连接
database.py
fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker DATABASE_URL="sqlite:///./test.db"engine=create_engine(DATABASE_URL,connect_args={"check_same_thread":False})SessionLocal=sessionmaker(autocommit=False,autoflush=False,bind=engine)依赖:获取 DB Session
from.databaseimportSessionLocaldefget_db():db=SessionLocal()try:yielddbfinally:db.close()使用 DB
fromfastapiimportDependsfromsqlalchemy.ormimportSession@app.get("/users")defget_users(db:Session=Depends(get_db)):users=db.query(User).all()returnusers获取当前用户
模拟 token 获取
fromfastapiimportHeader,HTTPExceptiondefget_token(authorization:str=Header(None)):ifnotauthorization:raiseHTTPException(status_code=401,detail="No token")# 简化:Bearer xxxtoken=authorization.replace("Bearer ","")returntoken模拟解析用户
# 这里依赖注入了get_tokendefget_current_user(token:str=Depends(get_token)):# 模拟 token 解析iftoken!="valid-token":raiseHTTPException(status_code=401,detail="Invalid token")return{"id":1,"username":"Tom"}嵌套依赖
在 FastAPI 中,依赖注入不仅可以用在接口函数上,还可以在依赖函数里继续依赖其他依赖,这就是嵌套依赖。
示例伪代码
最底层依赖:DB
defget_db():db="db_session"try:yielddbfinally:print("关闭 DB")中间层依赖:用户(依赖 DB)
fromfastapiimportDependsdefget_current_user(db=Depends(get_db)):return{"id":1,"username":"Tom","db":db}接口层(依赖用户)
@app.get("/profile")defprofile(user=Depends(get_current_user)):returnuser执行流程
当请求/profile时,执行顺序是:
profile ↓ get_current_user ↓ get_db ↓ 返回 db ↓ 返回 user ↓ 返回 response关于 FastAPI 的一些进阶功能(异常处理、中间件、依赖注入)。通过这些内容,我们已经可以让一个 FastAPI 项目具备基本的工程化能力。也希望对大家有所帮助。下一期我们再继续深入,聊一聊 FastAPI 项目的架构设计,比如如何分层、如何组织模块以及如何在实际项目中落地。