尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

零基础入门python51:FastAPI OAuth2表单登录与JWT

零基础入门python51:FastAPI OAuth2表单登录与JWT 零基础入门python51FastAPI OAuth2表单登录与JWT一、上一篇课后练习讲解上一篇练习围绕“注册与Argon2密码哈希”。参考做法是先运行上一篇的测试再用一个成功请求和一个失败请求验证边界本篇在同一项目上增加新能力。上一篇课后练习完整答案上一篇练习已经落实到完整文件运行下面代码可以观察本篇要求的成功和失败状态数据库写入全部放在明确事务边界内。答案要点注册接口校验密码长度并用 passlib 哈希数据库唯一约束处理并发重复邮箱异常 rollback 后返回 409。完整答案文件app/auth.py完整参考答案文件完整文件app/auth.pyfromfastapiimportAPIRouter,HTTPExceptionfrompasslib.contextimportCryptContextfrom.databaseimportget_db routerAPIRouter(prefix/auth)pwdCryptContext(schemes[bcrypt],deprecatedauto)router.post(/register)defregister(email:str,password:str):iflen(password)8:raiseHTTPException(400,password_too_short)return{email:email,password_hash:pwd.hash(password)}验收运行项目测试FastAPI/Flask 使用 python -m pytest -qDjango 使用 python manage.py test再按本文 curl 或 Docker 命令检查预期状态码。完整参考答案文件下面是交付项目中真实存在的完整文件 fastapi-blog/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))不要把哈希结果和用户输入直接比较也不要在日志中打印密码。验证失败时统一返回“邮箱或密码错误”避免泄露邮箱是否存在。二、本篇要解决的真实问题OAuth2PasswordRequestForm 读取表单账号密码JWT 只携带用户标识和过期时间不能把密码放进 token。三、请求是怎样走完整条链路的客户端请求 → 路由匹配 → Pydantic 校验 → Depends 注入用户/Session → SQLAlchemy 查询或业务服务 → 提交事务 → 响应模型序列化。每一步都有明确责任排错时按这个顺序定位。四、先看一个最小代码片段router.get(/api/health)defhealth():return{status:ok}这里的注释不是装饰它说明数据从哪里来、为什么不能相信客户端以及失败时系统应保持什么状态。五、完整文件代码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这五处合起来才是一个功能。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])阅读方法先找路由函数再找它的 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 何时提交、何时回滚。能用第二个用户验证资源隔离。能复现一个失败场景并说明原因。八、课后练习围绕“OAuth2表单登录与JWT”新增一个测试用例写出请求、预期响应和断言下一篇开头会给出参考实现并继续使用本项目。五、OAuth2 表单与 JWT登录成功不等于已经鉴权OAuth2PasswordRequestForm 读取 application/x-www-form-urlencoded不是 JSON。登录成功后签发短期 JWT之后请求使用 Authorization: Bearer token。router.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(status_code401,detail邮箱或密码错误)returnToken(access_tokencreate_token(user.id))令牌至少包含 sub 和 exp。sub 用字符串保存 id解码再转整数exp 让过期令牌自动失效。JWT 默认是签名不是加密不要塞密码或敏感信息。curl.exe-i-X POST http://127.0.0.1:8000/api/auth/login-HContent-Type: application/x-www-form-urlencoded--datausernamealiceexample.compasswordpassword123错误密码、缺少 Bearer 前缀、改动 token 一个字符都应 401。生产密钥来自环境变量换密钥会使旧 token 失效这是可控的安全操作。六、当前项目中的完整文件对照阅读七、令牌生命周期和撤销取舍短期 access token 减少泄露窗口但会增加重新登录次数若需要立即撤销可把 token 的 jti 放入 Redis 黑名单直到 exp。不要把长效 token 放在 URL 或日志中。客户端收到 401 后应清除本地 token 并重新登录不能无限重试造成请求风暴。面试常问“JWT 如何登出”答案不是删服务器上的 JWT而是短期 access token 加 refresh token 轮换或黑名单具体方案取决于安全要求和存储能力。八、补充代码文件本篇结束完整模块文件本节不是代码片段而是本篇结束时该模块的完整版本。请先备份旧文件再整体替换替换后重新运行本篇命令和测试。阅读时重点看本篇新增的函数、事务边界和错误处理未涉及的代码先不要自行删减。app/security.pyfromdatetimeimportdatetime,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])
返回列表