零基础入门python52FastAPI current_user鉴权依赖一、上一篇课后练习讲解上一篇练习围绕“OAuth2表单登录与JWT”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点create_access_token 写入 subject 与 UTC exp算法和 secret 集中配置令牌过期或篡改返回 401。完整答案文件app/security.py完整参考答案文件完整文件app/security.pyfromdatetimeimportdatetime,timedelta,timezonefromjoseimportjwt SECRETdev-change-medefcreate_access_token(subject:str,minutes:int30):expdatetime.now(timezone.utc)timedelta(minutesminutes)returnjwt.encode({sub:subject,exp:exp},SECRET,algorithmHS256)验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/app/security.py。它覆盖本篇新增逻辑以及前文已经完成的依赖代码复制单个函数会丢失上下文因此这里提供整份文件。fromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompwdlibimportPasswordHash SECRET_KEYdev-only-secret-key-change-in-env-123456789ALGORITHMHS256password_hashPasswordHash.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(minutes30)}returnjwt.encode(payload,SECRET_KEY,algorithmALGORITHM)defdecode_token(token):returnint(jwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])[sub])sub存用户 id 而不是邮箱邮箱可能修改exp必须由服务端生成不能相信客户端传入的过期时间。二、本篇要解决的真实问题鉴权拆成依赖后文章、评论、点赞都复用同一套 401 逻辑依赖还便于测试时 override。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段defroute(userDepends(get_current_user)):# user 由依赖从 JWT 解析客户端不能伪造 author_idreturn{user_id:user.id}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码app/dependencies.py当前阶段完整文件fromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjwtimportInvalidTokenErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.securityimportdecode_token oauth2_schemeOAuth2PasswordBearer(tokenUrl/api/auth/login)defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db)):try:user_iddecode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(401,无效或过期令牌)userdb.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 routerAPIRouter(prefix/api/auth,tags[auth])router.post(/register,response_modelUserRead,status_code201)defregister(body:UserCreate,db:SessionDepends(get_db)):ifdb.scalar(select(User).where(User.emailbody.email.lower())):raiseHTTPException(409,邮箱已注册)userUser(emailbody.email.lower(),password_hashhash_password(body.password))db.add(user);db.commit();db.refresh(user);returnuserrouter.post(/login,response_modelToken)deflogin(form:OAuth2PasswordRequestFormDepends(),db:SessionDepends(get_db)):userdb.scalar(select(User).where(User.emailform.username.lower()))ifnotuserornotverify_password(form.password,user.password_hash):raiseHTTPException(401,邮箱或密码错误)returnToken(access_tokencreate_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_schemeOAuth2PasswordBearer(tokenUrl/api/auth/login)defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db)):try:user_iddecode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(status_code401,detail无效或过期令牌)userdb.get(User,user_id)ifnotuser:raiseHTTPException(status_code401,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.pyfromfastapiimportDepends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjwtimportInvalidTokenErrorfromsqlalchemy.ormimportSessionfrom.databaseimportget_dbfrom.modelsimportUserfrom.securityimportdecode_token oauth2_schemeOAuth2PasswordBearer(tokenUrl/api/auth/login)defget_current_user(token:strDepends(oauth2_scheme),db:SessionDepends(get_db)):try:user_iddecode_token(token)except(InvalidTokenError,ValueError):raiseHTTPException(401,无效或过期令牌)userdb.get(User,user_id)ifnotuser:raiseHTTPException(401,用户不存在)returnuser