FastAPI 依赖注入实战:鉴权、数据库会话与复用
FastAPI 依赖注入实战:鉴权、数据库会话与复用用 FastAPI 写着写着你会发现:几乎每个接口开头都在重复同一套动作——校验 token、开数据库会话、读当前用户。复制粘贴到第十个接口,你就该问自己:这些前置逻辑能不能抽出来?FastAPI 的Depends就是干这个的。这篇不背概念,直接用鉴权和数据库两个最高频的场景,把依赖注入讲到能上手。从重复代码说起先看没有依赖注入时,接口有多啰嗦:fromfastapiimportFastAPI,HTTPException,Header appFastAPI()app.get(/profile)defprofile(authorization:strHeader(None)):# 每个需要登录的接口都要抄这一段ifnotauthorization:raiseHTTPException(401,缺少 token)tokenauthorization.replace(Bearer ,)userdecode_token(token)# 假设这函数解出用户ifnotuser:raiseHTTPException(401,token 无效)return{name:user[name]}这段鉴权逻辑,在/orders、/settings、/messages里会一字不差再出现。改一次校验规则,你得改十个地方。依赖注入要解决的就是这个。第一个依赖:把鉴权抽出来Depends的用法是:把一段前置逻辑写成函数,接口参数里声明依赖它,FastAPI 会在调用接口前自动执行这个函数,把返回值塞给参数。fromfastapiimportDepends,HTTPException,Headerdefget_current_user(authorization:strHeader(None)):ifnotauthorization:raiseHTTPException(401,缺少 token)tokenauthorization.replace(Bearer ,)userdecode_token(token)ifnotuser:raiseHTTPException(401,token 无效)returnuser# 返回值会成为依赖它的接口参数app.get(/profile)defprofile(user:dictDepends(get_current_user)):# 进到这里 user 一定是合法登录用户,鉴权逻辑一行都不用写return{name:user[name]}app.get(/orders)deforders(user:dictDepends(get_current_user)):return{orders:load_orders(user[id])}现在鉴权只写一遍。/profile和/orders只要声明Depends(get_current_user),进入函数体时user必然是合法的——校验失败早在依赖里就抛 401 了,接口逻辑保持干净。第二个依赖:数据库会话与 yield数据库会话有个特点:用完必须关。FastAPI 的依赖支持用yield,让你在请求处理前后各做一件事——请求前开会话,请求后关会话,哪怕接口报错也保证关闭。fromsqlalchemy.ormimportSessiondefget_db():dbSessionLocal()# 请求进来:开一个会话try:yielddb# 把会话交给接口用finally:db.close()# 请求结束(无论成功报错):一定关闭app.get(/users/{uid})defget_user(uid:int,db:SessionDepends(get_db)):userdb.query(User).filter(User.iduid).first()ifnotuser:raiseHTTPException(404,用户不存在)returnuseryield之前的代码在请求开始时跑,yield之后的在响应发出后跑。finally保证了会话不泄漏——这是手写 try/finally 最容易忘、依赖注入帮你兜住的地方。依赖可以嵌套:组合出真实业务真实场景往往是「先有数据库会话,再基于会话查当前用户」。依赖之间可以互相依赖,FastAPI 会自动理顺执行顺序。defget_current_user(authorization:strHeader(None),db:SessionDepends(get_db),# 依赖里再依赖数据库会话):ifnotauthorization:raiseHTTPException(401,缺少 token)tokenauthorization.replace(Bearer ,)payloaddecode_token(token)ifnotpayload:raiseHTTPException(401,token 无效)# 用会话去库里查真实用户,而不是只信 token 里的内容userdb.query(User).filter(User.idpayload[id]).first()ifnotuser:raiseHTTPException(401,用户已注销)returnuserapp.get(/me)defme(user:UserDepends(get_current_user)):# get_db 和 get_current_user 都被自动执行,顺序 FastAPI 排好return{id:user.id,name:user.name}调/me时,FastAPI 先跑get_db拿到会话,再把会话传给get_current_user,最后把查到的user给接口。你只声明依赖关系,执行顺序不用操心。进阶:参数化依赖做权限分级如果不同接口要求不同角色(普通用户 vs 管理员),可以写一个「返回依赖的工厂」,用类实现,带上参数。classRequireRole:def__init__(self,role:str):self.rolerole# 记住需要的角色def__call__(self,user:UserDepends(get_current_user)):# 类的实例可像函数一样被 Depends 调用ifuser.role!self.role:raiseHTTPException(403,f需要{self.role}权限)returnuserapp.get(/admin/stats)defstats(user:UserDepends(RequireRole(admin))):# 只有 admin 能进,权限判断收敛在 RequireRole 里return{total_users:12345}RequireRole(admin)生成一个可调用对象,Depends会调用它并自动注入get_current_user的结果。同一套机制,你能派生出RequireRole(vip)等各种权限门,接口层只需换一个参数。一个常被忽略的收益:测试时轻松替换依赖依赖注入还有个隐藏福利:测试时可以用dependency_overrides把真实依赖换成假的,不用连真数据库、不用造真 token。deffake_user():returnUser(id1,nametest,roleadmin)# 测试里把 get_current_user 换成假的,直接跳过鉴权app.dependency_overrides[get_current_user]fake_user这一行让所有依赖get_current_user的接口在测试中都拿到假用户,测试从此不依赖外部环境。生产代码不用改,只在测试里覆盖——这是依赖注入相比硬编码最实用的好处之一。小结依赖注入解决的痛点:把鉴权、开会话这类每个接口都要做的前置逻辑抽成一处,改一次全局生效。Depends(func):接口调用前自动执行func,返回值注入参数;校验失败在依赖里抛异常,接口体保持干净。yield型依赖:yield前做准备(开会话),finally里做清理(关会话),异常也能保证释放资源。依赖可嵌套、可参数化:组合出「会话→当前用户→角色校验」的真实链路;用类 __call__做权限分级。测试友好:dependency_overrides一行替换依赖,测试不碰真实数据库和 token。一句话记忆:凡是「每个接口开头都要抄」的逻辑,就该写成Depends。