Python全栈实战 Day 3用户注册与JWT登录认证前言这是“10天Python全栈项目实战”系列的第3篇。前两天已经完成FastAPI与Vue3项目搭建第一个前后端接口调用MySQL数据库连接SQLAlchemy用户模型创建。今天将在现有项目中实现完整的后端用户认证流程用户注册密码哈希用户登录生成JWT Token根据Token获取当前用户。注意数据库中不能直接保存用户明文密码只保存经过密码哈希算法处理后的结果。一、安装认证依赖进入后端项目目录并激活虚拟环境cd backend .venv\Scripts\activate安装依赖pip install pyjwt pwdlib[argon2] python-dotenv email-validator各依赖的作用依赖作用PyJWT生成和解析JWTpwdlib密码哈希与校验argon2实际使用的密码哈希算法python-dotenv读取环境变量email-validator校验邮箱格式FastAPI当前官方JWT教程使用PyJWT处理Token并使用pwdlib处理密码。PasswordHash.recommended()目前默认选择Argon2。更新依赖文件pip freeze requirements.txt二、调整项目结构本文按照以下后端结构编写backend ├── app │ ├── api │ │ ├── __init__.py │ │ └── auth.py │ ├── core │ │ ├── __init__.py │ │ └── security.py │ ├── models │ │ ├── __init__.py │ │ └── user.py │ ├── schemas │ │ ├── __init__.py │ │ └── user.py │ ├── __init__.py │ ├── database.py │ └── main.py ├── .env ├── .gitignore └── requirements.txtdatabase.py和models/user.py应该已经在Day 2中创建。三、检查用户数据模型打开app/models/user.py确认用户表至少包含以下字段from datetime import datetime from sqlalchemy import Boolean, DateTime, String, func from sqlalchemy.orm import Mapped, mapped_column from app.database import Base class User(Base): __tablename__ users id: Mapped[int] mapped_column( primary_keyTrue, indexTrue, ) username: Mapped[str] mapped_column( String(30), uniqueTrue, indexTrue, nullableFalse, ) email: Mapped[str] mapped_column( String(100), uniqueTrue, indexTrue, nullableFalse, ) hashed_password: Mapped[str] mapped_column( String(255), nullableFalse, ) is_active: Mapped[bool] mapped_column( Boolean, defaultTrue, nullableFalse, ) created_at: Mapped[datetime] mapped_column( DateTime, server_defaultfunc.now(), nullableFalse, )这里保存的是hashed_password而不是password如果Day 2创建的表没有hashed_password字段需要修改数据模型和数据库表。开发阶段没有重要数据时可以删除旧表后重新创建已有数据时应使用数据库迁移或执行对应的ALTER TABLE语句不能直接删表。四、创建请求和响应模型创建app/schemas/user.py写入from pydantic import BaseModel, ConfigDict, EmailStr, Field class UserRegister(BaseModel): username: str Field( min_length3, max_length30, patternr^[a-zA-Z0-9_]$, ) email: EmailStr password: str Field( min_length8, max_length128, ) class UserLogin(BaseModel): username: str password: str class TokenResponse(BaseModel): access_token: str token_type: str bearer class UserResponse(BaseModel): id: int username: str email: EmailStr is_active: bool model_config ConfigDict(from_attributesTrue)这几个模型分别负责UserRegister校验注册参数UserLogin校验登录参数TokenResponse定义登录成功的返回内容UserResponse定义返回给前端的用户信息。不要把hashed_password放进UserResponse否则可能将密码哈希结果返回给前端。五、配置JWT密钥在项目根目录创建.env先生成随机密钥python -c import secrets; print(secrets.token_hex(32))终端会输出一段随机字符串将它复制进.envSECRET_KEY替换成刚才生成的随机字符串 ACCESS_TOKEN_EXPIRE_MINUTES30创建或检查.gitignore.venv/ __pycache__/ *.pyc .env.env中包含JWT签名密钥不应该提交到公开代码仓库。六、实现密码加密和JWT生成创建app/core/security.py写入import os from datetime import datetime, timedelta, timezone import jwt from dotenv import load_dotenv from pwdlib import PasswordHash load_dotenv() SECRET_KEY os.getenv(SECRET_KEY) ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES int( os.getenv(ACCESS_TOKEN_EXPIRE_MINUTES, 30) ) if not SECRET_KEY: raise RuntimeError(没有配置SECRET_KEY请检查.env文件) password_hash PasswordHash.recommended() def hash_password(password: str) - str: 将明文密码转换为密码哈希。 return password_hash.hash(password) def verify_password( plain_password: str, hashed_password: str, ) - bool: 验证明文密码是否与密码哈希匹配。 return password_hash.verify( plain_password, hashed_password, ) def create_access_token( subject: str, expires_delta: timedelta | None None, ) - str: 生成JWT访问令牌。 expire datetime.now(timezone.utc) ( expires_delta if expires_delta is not None else timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) ) payload { sub: subject, exp: expire, } return jwt.encode( payload, SECRET_KEY, algorithmALGORITHM, ) def decode_access_token(token: str) - dict: 校验并解析JWT访问令牌。 return jwt.decode( token, SECRET_KEY, algorithms[ALGORITHM], )JWT中保存{ sub: 用户ID, exp: 过期时间, }其中sub表示Token对应的主体exp表示Token过期时间。PyJWT在执行jwt.decode()时会校验expToken过期后会抛出相应异常解析时还应显式限制允许使用的签名算法。七、实现注册接口创建app/api/auth.py先写入注册功能import jwt from fastapi import ( APIRouter, Depends, HTTPException, status, ) from fastapi.security import ( HTTPAuthorizationCredentials, HTTPBearer, ) from sqlalchemy import select from sqlalchemy.orm import Session from app.core.security import ( create_access_token, decode_access_token, hash_password, verify_password, ) from app.database import get_db from app.models.user import User from app.schemas.user import ( TokenResponse, UserLogin, UserRegister, UserResponse, ) router APIRouter( prefix/api/auth, tags[用户认证], ) bearer_scheme HTTPBearer(auto_errorFalse) router.post( /register, response_modelUserResponse, status_codestatus.HTTP_201_CREATED, ) def register( data: UserRegister, db: Session Depends(get_db), ) - User: username_exists db.scalar( select(User).where( User.username data.username ) ) if username_exists: raise HTTPException( status_codestatus.HTTP_409_CONFLICT, detail用户名已经存在, ) email_exists db.scalar( select(User).where( User.email data.email ) ) if email_exists: raise HTTPException( status_codestatus.HTTP_409_CONFLICT, detail邮箱已经被注册, ) user User( usernamedata.username, emaildata.email, hashed_passwordhash_password(data.password), ) try: db.add(user) db.commit() db.refresh(user) except Exception: db.rollback() raise HTTPException( status_codestatus.HTTP_500_INTERNAL_SERVER_ERROR, detail用户注册失败, ) return user注册接口的主要流程接收注册参数 ↓ 检查用户名是否存在 ↓ 检查邮箱是否存在 ↓ 对密码执行哈希 ↓ 写入MySQL ↓ 返回用户基本信息八、实现登录接口继续在auth.py中加入router.post( /login, response_modelTokenResponse, ) def login( data: UserLogin, db: Session Depends(get_db), ) - TokenResponse: user db.scalar( select(User).where( User.username data.username ) ) if user is None: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误, ) if not verify_password( data.password, user.hashed_password, ): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户名或密码错误, ) if not user.is_active: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detail当前账号已被禁用, ) access_token create_access_token( subjectstr(user.id) ) return TokenResponse( access_tokenaccess_token, )无论用户名不存在还是密码错误都统一返回用户名或密码错误这样可以减少接口向外部暴露账号是否存在的信息。登录成功后返回{ access_token: 生成的JWT字符串, token_type: bearer }九、获取当前登录用户继续加入Token验证依赖def get_current_user( credentials: HTTPAuthorizationCredentials | None Depends( bearer_scheme ), db: Session Depends(get_db), ) - User: if credentials is None: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail请先登录, headers{WWW-Authenticate: Bearer}, ) try: payload decode_access_token( credentials.credentials ) user_id payload.get(sub) if user_id is None: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail无效的登录凭证, ) user_id_int int(user_id) except ( jwt.InvalidTokenError, ValueError, TypeError, ): raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail登录凭证无效或已经过期, headers{WWW-Authenticate: Bearer}, ) user db.get(User, user_id_int) if user is None: raise HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detail用户不存在, ) if not user.is_active: raise HTTPException( status_codestatus.HTTP_403_FORBIDDEN, detail当前账号已被禁用, ) return user然后创建受保护接口router.get( /me, response_modelUserResponse, ) def get_me( current_user: User Depends(get_current_user), ) - User: return current_user访问/api/auth/me时必须携带Authorization: Bearer JWT_TOKEN否则接口会返回401。十、注册认证路由打开app/main.py加入认证路由from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.auth import router as auth_router app FastAPI( title10天Python全栈项目实战, version1.0.0, ) app.add_middleware( CORSMiddleware, allow_origins[ http://localhost:5173, http://127.0.0.1:5173, ], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) app.include_router(auth_router) app.get(/) def read_root() - dict[str, str]: return { message: FastAPI服务运行成功, }启动项目uvicorn app.main:app --reload打开接口文档http://127.0.0.1:8000/docs十一、测试注册接口找到POST /api/auth/register测试参数{ username: owen, email: owenexample.com, password: Python123456 }成功后返回{ id: 1, username: owen, email: owenexample.com, is_active: true }建议在这里放两张截图Swagger注册成功MySQL用户表新增数据。查看数据库时hashed_password应该是一段密码哈希而不是Python123456十二、测试登录接口找到POST /api/auth/login提交{ username: owen, password: Python123456 }成功后获得{ access_token: 一段JWT字符串, token_type: bearer }复制access_token备用。十三、测试登录用户接口找到GET /api/auth/me点击Swagger页面右上角的“Authorize”输入Bearer 你的JWT_TOKEN部分Swagger页面只需要填写Token本身不需要手动写Bearer以页面提示为准。认证成功后执行GET /api/auth/me应该返回{ id: 1, username: owen, email: owenexample.com, is_active: true }再测试以下情况不携带Token随便修改Token等Token过期后再次请求。这些情况都应该返回401而不是正常获取用户信息。十四、常见报错1. 提示找不到jwt执行pip install pyjwt注意安装包名称是PyJWT代码导入方式是import jwt2. 提示需要安装email-validator执行pip install email-validator因为Pydantic的EmailStr依赖该组件验证邮箱。3. 提示没有配置SECRET_KEY检查.env是否位于后端项目根目录backend/.env并确认内容为SECRET_KEY真实生成的随机密钥修改.env后重新启动FastAPI。4. 用户表没有hashed_password字段说明数据库结构仍然是Day 2的旧版本。需要更新表结构或在没有重要数据的开发环境中删除旧表后重新创建。5. 正确密码也登录失败重点检查注册时是否调用了hash_password()数据库保存的是否为哈希结果登录时是否使用verify_password()是否手动修改过数据库中的密码字段。6. Token刚生成就提示无效重点检查生成和解析Token是否使用同一个SECRET_KEY使用的算法是否都是HS256是否重启项目后更换了随机密钥.env是否正确加载。十五、Day 3验收清单完成以下检查说明Day 3已经通过注册接口能够创建用户用户名不能重复邮箱不能重复密码少于8位时校验失败数据库没有保存明文密码正确账号密码可以登录错误密码无法登录登录成功可以获得JWT Token携带Token可以访问/api/auth/me不携带Token时返回401伪造或过期Token无法通过认证十六、下一篇内容Day 4将完成Vue3登录页面、Axios请求封装、Token保存和路由守卫。下一篇主要包括创建登录页面调用FastAPI登录接口保存JWT TokenAxios自动携带Token未登录用户自动跳转登录页登录后获取当前用户信息。
