零基础入门python52：FastAPI current_user鉴权依赖

零基础入门python52：FastAPI current_user鉴权依赖

一、上一篇课后练习讲解

上一篇练习围绕“OAuth2表单登录与JWT”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：create_access_token 写入 subject 与 UTC exp，算法和 secret 集中配置；令牌过期或篡改返回 401。

完整答案文件：app/security.py

完整参考答案文件

完整文件：app/security.py

fromdatetimeimportdatetime,timedelta,timezonefromjoseimportjwt SECRET="dev-change-me"defcreate_access_token(subject:str,minutes:int=30):exp=datetime.now(timezone.utc)+timedelta(minutes=minutes)returnjwt.encode({"sub":subject,"exp":exp},SECRET,algorithm="HS256")

验收：运行项目测试（FastAPI/Flask 使用 python -m pytest -q，Django 使用 python manage.py test），再按本文 curl 或 Docker 命令检查预期状态码。

完整参考答案文件

下面是交付项目中真实存在的完整文件 fastapi-blog/app/security.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码；复制单个函数会丢失上下文，因此这里提供整份文件。

fromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompwdlibimportPasswordHash SECRET_KEY='dev-only-secret-key-change-in-env-123456789'ALGORITHM='HS256'password_hash=PasswordHash.recommended()defhash_password(password):returnpassword_hash.hash(password)defverify_password(password,hashed):returnpassword_hash.verify(password,hashed)defcreate_token(user_id):payload={'sub':str(user_id),'exp':datetime.now(timezone.utc)+timedelta(minutes=30)}returnjwt.encode(payload,SECRET_KEY,algorithm=ALGORITHM)defdecode_token(token):returnint(jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM])['sub'])

sub存用户 id 而不是邮箱，邮箱可能修改；exp必须由服务端生成，不能相信客户端传入的过期时间。

二、本篇要解决的真实问题

鉴权拆成依赖后，文章、评论、点赞都复用同一套 401 逻辑；依赖还便于测试时 override。

三、请求是怎样走完整条链路的

客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任，排错时按这个顺序定位。

四、先看一个最小代码片段

defroute(user=Depends(get_current_user)):# user 由依赖从 JWT 解析，客户端不能伪造 author_idreturn{"user_id":user.id}

这里的注释不是装饰：它说明数据从哪里来、为什么不能相信客户端，以及失败时系统应保持什么状态。

五、完整文件代码

app/dependencies.py（当前阶段完整文件）

fromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjwtimportInvalidTokenErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.securityimportdecode_token oauth2_scheme=OAuth2PasswordBearer(tokenUrl='/api/auth/login')defget_current_user(token:str=Depends(oauth2_scheme),db:Session=Depends(get_db)):try:user_id=decode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(401,'无效或过期令牌')user=db.get(User,user_id)ifnotuser:raiseHTTPException(401,'用户不存在')returnuser

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

app/auth.py（当前阶段完整文件）

fromfastapiimportAPIRouter,Depends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordRequestFormfromsqlalchemyimportselectfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.schemasimportToken,UserCreate,UserReadfrom.securityimportcreate_token,hash_password,verify_password router=APIRouter(prefix='/api/auth',tags=['auth'])@router.post('/register',response_model=UserRead,status_code=201)defregister(body:UserCreate,db:Session=Depends(get_db)):ifdb.scalar(select(User).where(User.email==body.email.lower())):raiseHTTPException(409,'邮箱已注册')user=User(email=body.email.lower(),password_hash=hash_password(body.password))db.add(user);db.commit();db.refresh(user);returnuser@router.post('/login',response_model=Token)deflogin(form:OAuth2PasswordRequestForm=Depends(),db:Session=Depends(get_db)):user=db.scalar(select(User).where(User.email==form.username.lower()))ifnotuserornotverify_password(form.password,user.password_hash):raiseHTTPException(401,'邮箱或密码错误')returnToken(access_token=create_token(user.id))

阅读方法：先找路由函数，再找它的 Depends、输入 schema、数据库操作和 response_model；这五处合起来才是一个功能。

六、安装、启动与验收

..\..\.venv\Scripts\python.exe-m pip install-r requirements.txt..\..\.venv\Scripts\python.exe-m pytest-q..\..\.venv\Scripts\python.exe-m uvicorn app.main:app--reload

打开/docs，按顺序完成注册、登录、创建文章、分页查询、修改删除、评论点赞。错误请求必须看到明确状态码：校验错误 422，未登录 401，无权限 403，重复点赞 409。测试应全部通过。

七、本篇验收清单

• 能指出输入校验发生在哪个 schema。
• 能解释 Session 何时提交、何时回滚。
• 能用第二个用户验证资源隔离。
• 能复现一个失败场景并说明原因。

八、课后练习

围绕“current_user鉴权依赖”新增一个测试用例，写出请求、预期响应和断言；下一篇开头会给出参考实现，并继续使用本项目。

五、current_user 是可复用的门卫

身份认证和资源授权是两个问题：current_user 只回答“你是谁”，文章路由再回答“你能修改哪篇文章”。分开后，管理员规则和作者规则才能独立演进。

oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/api/auth/login")defget_current_user(token:str=Depends(oauth2_scheme),db:Session=Depends(get_db)):try:user_id=decode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(status_code=401,detail="无效或过期令牌")user=db.get(User,user_id)ifnotuser:raiseHTTPException(status_code=401,detail="用户不存在")returnuser

验收覆盖三种结果：不带 token 是 401；另一个用户修改资源是 403；随机 id 是 404。若三种都变成 403，说明把认证和授权混在一起。删除用户后，旧 token 即使签名有效，也应因查不到用户而 401。

六、当前项目中的完整文件（对照阅读）

七、授权检查的数据库写法

除了“先查再判断”，还可以把 owner_id 直接放进 UPDATE 条件，返回受影响行数为 0 时再区分 404 和 403。这种写法减少竞态窗口，但查询和错误语义必须清楚。管理员权限应来自数据库角色，不应相信客户端传来的 is_admin 字段。

审计中记录 user_id、article_id、动作和结果。即使返回 403，也不要把请求体中的正文写进日志，避免敏感内容泄露。

八、补充代码文件

本篇结束：完整模块文件

本节不是代码片段，而是本篇结束时该模块的完整版本。请先备份旧文件，再整体替换；替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理，未涉及的代码先不要自行删减。

app/dependencies.py

fromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjwtimportInvalidTokenErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.securityimportdecode_token oauth2_scheme=OAuth2PasswordBearer(tokenUrl='/api/auth/login')defget_current_user(token:str=Depends(oauth2_scheme),db:Session=Depends(get_db)):try:user_id=decode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(401,'无效或过期令牌')user=db.get(User,user_id)ifnotuser:raiseHTTPException(401,'用户不存在')returnuser