企业官网建设流程全解析

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

一、上一篇课后练习讲解

上一篇练习围绕“注册与Argon2密码哈希”。参考做法是先运行上一篇的测试，再用一个成功请求和一个失败请求验证边界；本篇在同一项目上增加新能力。

上一篇课后练习完整答案

上一篇练习已经落实到完整文件，运行下面代码可以观察本篇要求的成功和失败状态；数据库写入全部放在明确事务边界内。
答案要点：注册接口校验密码长度并用 passlib 哈希，数据库唯一约束处理并发重复邮箱，异常 rollback 后返回 409。

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

完整参考答案文件

完整文件：app/auth.py

fromfastapiimportAPIRouter,HTTPExceptionfrompasslib.contextimportCryptContextfrom.databaseimportget_db router=APIRouter(prefix="/auth")pwd=CryptContext(schemes=["bcrypt"],deprecated="auto")@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 -q，Django 使用 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 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))

不要把哈希结果和用户输入直接比较，也不要在日志中打印密码。验证失败时统一返回“邮箱或密码错误”，避免泄露邮箱是否存在。

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

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 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；这五处合起来才是一个功能。

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'])

阅读方法：先找路由函数，再找它的 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_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(status_code=401,detail="邮箱或密码错误")returnToken(access_token=create_token(user.id))

令牌至少包含 sub 和 exp。sub 用字符串保存 id，解码再转整数；exp 让过期令牌自动失效。JWT 默认是签名不是加密，不要塞密码或敏感信息。

curl.exe-i-X POST http://127.0.0.1:8000/api/auth/login-H"Content-Type: application/x-www-form-urlencoded"--data"username=alice@example.com&password=password123"

错误密码、缺少 Bearer 前缀、改动 token 一个字符都应 401。生产密钥来自环境变量，换密钥会使旧 token 失效，这是可控的安全操作。

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

七、令牌生命周期和撤销取舍

短期 access token 减少泄露窗口，但会增加重新登录次数；若需要立即撤销，可把 token 的 jti 放入 Redis 黑名单直到 exp。不要把长效 token 放在 URL 或日志中。客户端收到 401 后应清除本地 token 并重新登录，不能无限重试造成请求风暴。

面试常问“JWT 如何登出”：答案不是删服务器上的 JWT，而是短期 access token 加 refresh token 轮换或黑名单，具体方案取决于安全要求和存储能力。

八、补充代码文件

本篇结束：完整模块文件

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

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'])