《FastAPI + SQLAlchemy异步实战（二）：构建用户接口》

《FastAPI + SQLAlchemy异步实战（二）：构建用户接口》

此部分讲解对用户功能接口的讲解，相信对于大部分后端项目来说，用户功能相关接口一定都是必不可少的，相较于新闻接口这部分难度更大，涉及用户注册密码、验证token、更改用户信息等等。让我们一起走进用户登录接口的大门吧。

一、用户注册接口

用户登录接口的主要要求是上传用户名以及密码（加密），由此创建一个新用户，并在注册用户时创建一个token以及其终止时间，其目的是：不用每次请求都传用户名 + 密码，保护账号密码。如果每次访问接口都把账号密码放在请求里：

• 密码反复在网络传输，泄露风险高；
• 前端需要一直保存用户密码，前端很容易被窃取（JS 泄露、抓包）。

有了token之后整个登录流程就变为了：

1. 登录接口：提交账号密码，校验正确 → 服务器生成Token 返回给前端
2. 之后所有业务接口：前端只携带 Token，不再传账号密码服务器解析 Token 就知道 “这是谁在访问”。

1.接口核心代码：

@router.post("/register")asyncdefregister(user_data:UserRequest,db:AsyncSession=Depends(get_db)):existing_user=awaitusers.get_user(db,user_data.username)ifexisting_user:# 在注册用户之前，必须先判断用户是否已经存在。raiseHTTPException(status_code=400,detail="用户已存在")user=awaitusers.create_user(db,user_data)token=awaitusers.create_token(db,user.id)response_data=UserAuthResponse(token=token,user_info=UserInfoResponse.model_validate(user))returnsuccess_response(message="注册成功",data=response_data)# 用户注册提交表单，用的是请求体参数，需要Pydantic校验，上传用户所需的用户名和密码classUserRequest(BaseModel):username:strpassword:str# 根据用户名查询数据库asyncdefget_user(db:AsyncSession,username:str):query=select(User).where(User.username==username)result=awaitdb.execute(query)returnresult.scalar_one_or_none()# 创建用户asyncdefcreate_user(db:AsyncSession,user_data:UserRequest):# 先对密码加密处理→addhashed_password=get_hash_password(user_data.password)user=User(username=user_data.username,password=hashed_password)db.add(user)awaitdb.commit()awaitdb.refresh(user)# 刷新数据(从数据库中获取最新的user)returnuser# 为了保护用户数据安全，在用户密码上传到数据库时，必须进行加密处理。frompasslib.contextimportCryptContext# 创建密码上下文对象，指定使用的加密算法为 bcrypt，并设置过时的算法为自动处理pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto")# 密码加密defget_hash_password(password:str):returnpwd_context.hash(password)# 密码校验defverify_password(plain_password:str,hashed_password:str)->bool:# 指明密码返回结果是bool类型。returnpwd_context.verify(plain_password,hashed_password)# 拿数据库中的密码和用户输入的密码进行对比验证（用数据库里面密码的盐加明文密码再次加密比对）# 生成token：Token 就是一张 “临时身份证”。登录时核对账号密码，发你身份证；之后办事不用再出示账号密码，出示这张身份证就可以证明你是谁。身份证有有效期，到期作废。asyncdefcreate_token(db:AsyncSession,user_id:int):# 创建token+设置过期时间+当前用户是否有权限（有更新，没有则添加）token=secrets.token_urlsafe(32)expire_time=datetime.now()+timedelta(days=7)query=select(UserToken).where(UserToken.user_id==user_id)result=awaitdb.execute(query)user_token=result.scalar_one_or_none()ifuser_token:# 更新user_token.token=token user_token.expires_at=expire_timeelse:# 添加user_token=UserToken(user_id=user_id,token=token,expires_at=expire_time)db.add(user_token)awaitdb.commit()returntoken

用户信息数据的封装：为了简化接口响应json格式复杂的填写，提前吧用户数据封好，在接口retuen，直接调用。

classUserInfoBase(BaseModel):""" 用户基础信息 """nickname:Optional[str]=Field(None,max_length=50,description="昵称")avatar:Optional[str]=Field(None,max_length=255,description="头像")gender:Optional[str]=Field(None,max_length=10,description="性别")bio:Optional[str]=Field(None,max_length=500,description="个人简介")classUserInfoResponse(UserInfoBase):id:intusername:str# 模型封裝配置model_config=ConfigDict(from_attributes=True,# 允许从ORM对象属性中填充数据)# data要有自己的类型classUserAuthResponse(BaseModel):token:struser_info:UserInfoResponse=Field(...,alias="userInfo")# 模型封裝配置model_config=ConfigDict(populate_by_name=True,# 允许使用别名进行数据填充/字段名兼容from_attributes=True,# 允许从ORM对象属性中填充数据)

2.两种token使用方式对比

本项目create_token 代码逻辑拆解（数据库存储随机 token）

1. token 本身只是一串随机字符串，不携带任何业务信息，本身看不出user_id、权限、过期时间；
2. 所有信息全部保存在数据库UserToken表：user_id、token字符串、expires_at过期时间；
3. 校验流程：

客户端请求头带上Authorization: Bearer {token}

1. 后端拿到 token，去数据库查询 UserToken 表
2. 判断：token 是否存在、是否没过期、用户是否有效
3. 通过才拿到user_id，鉴权放行

4. 的业务逻辑：同一个用户只会保留一条 token 记录，调用create_token直接覆盖旧 token，旧 token 立刻失效，天然实现单点登录。

JWT方法

1. 把用户身份、过期时间等信息编码放在 Token 字符串本身，配合签名防篡改，默认不需要查数据库就能完成鉴权。

2. 格式三段，用.隔开：header.payload.signature

   • header（头部）：签名算法，如 HS256

   • payload（载荷）：业务数据user_id、exp(过期时间)，只是 Base64 编码，不是加密，任何人都能解码看到内容，不能存密码

   • signature（签名）：用密钥对前两段做哈希，防止篡改；密钥只保存在服务端。

   篡改 header/payload，签名校验直接失败，拒绝请求。

3. 完整工作流程

   1. 登录：账号密码校验通过 → 组装 payload → 用密钥生成 JWT 返回客户端
   2. 请求接口：客户端请求头携带Authorization: Bearer jwt字符串
   3. 校验：后端校验签名 + 校验 exp 过期时间 → 解析出 user_id，直接鉴权，不用查库
   4. 退出登录痛点：原生 JWT 无法主动作废，没到过期时间就一直有效。

4. 优缺点

✅ 优点

1. 无状态：服务端不用存会话数据，适合分布式、微服务，多台机器直接共用
2. 鉴权快，省去数据库 / Redis 查询
3. 适合 APP、小程序

❌ 缺点

1. 不能主动失效：改密码、踢下线、登出，JWT 还会生效直到 exp 到期；要失效就得做 Redis 黑名单，又变回需要存储，丢掉无状态优势
2. token 字符串长，每次 http 请求头体积更大
3. payload 公开可见，敏感数据不能放
4. 不建议设置长有效期，长时效 JWT 泄露风险巨大

fromdatetimeimportdatetime,timedeltafromtypingimportOptionalimportjwtfrompydanticimportBaseModel# 生产环境务必放到环境变量，不要硬编码！SECRET_KEY="your-strong-secret-key-keep-it-safe"ALGORITHM="HS256"ACCESS_TOKEN_EXPIRE_MINUTES=15# access_token短时效15分钟classTokenData(BaseModel):user_id:Optional[int]=Nonedefcreate_access_token(user_id:int)->str:"""生成JWT access_token"""expire=datetime.utcnow()+timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)payload={"sub":str(user_id),# sub 标准字段，存用户id，转字符串是jwt惯例"exp":expire# 过期时间戳，jwt会自动校验}token=jwt.encode(payload,SECRET_KEY,algorithm=ALGORITHM)returntokendefverify_access_token(token:str)->TokenData:"""校验jwt，解析用户信息，签名错误/过期直接抛异常"""try:payload=jwt.decode(token,SECRET_KEY,algorithms=[ALGORITHM],options={"verify_exp":True})user_id:str=payload.get("sub")ifuser_idisNone:raisejwt.InvalidTokenError("token无用户标识")returnTokenData(user_id=int(user_id))exceptjwt.ExpiredSignatureError:raiseException("token已过期")exceptjwt.InvalidTokenError:raiseException("token非法，签名校验失败")

二、用户登录接口（login）

1.接口核心代码

@router.post("/login")asyncdeflogin(user_data:UserRequest,db:AsyncSession=Depends(get_db)):user=awaitusers.get_user(db,user_data.username)ifnotuserornotverify_password(user_data.password,user.password):# 验证密码是否正确(明文密码与数据库中加密后的密码进行比对)raiseHTTPException(status_code=400,detail="用户名或密码错误")token=awaitusers.create_token(db,user.id)response_data=UserAuthResponse(token=token,user_info=UserInfoResponse.model_validate(user))returnsuccess_response(message="登录成功",data=response_data)

这里的user是crud文件夹下get_user方法获取的ORM类型的数据，其包含了数据表中所有的数据类型，所以在传入UserInfoResponse这个pydantic这个类的时候，需要进行model_validate()，这样可以自动筛选出这个类别所需的对象。

三、用户信息相关接口（获取、修改）

1.接口核心代码

# 获取用户信息接口→封装CRUD→功能整合成一个工具函数→路由导入使用：注入依赖@router.get("/info")asyncdefget_user_info(user=Depends(get_current_user)):# 此处没有db是因为不用调用数据库进行CRUD，仅仅要验证token拿到用户信息returnsuccess_response(message="获取用户信息成功",data=UserInfoResponse.model_validate(user))# 修改用户信息接口:验证token→更新（用户输入数据put）→返回成功响应@router.put("/update")asyncdefupdate_user_info(user_data:UserUpdateRequest,user=Depends(get_current_user),db:AsyncSession=Depends(get_db)):user=awaitusers.update_user(db,user.username,user_data)returnsuccess_response(message="更新用户信息成功",data=UserInfoResponse.model_validate(user))# 更新用户的crud函数如下：asyncdefupdate_user(db:AsyncSession,username:str,user_data:UserUpdateRequest):# user_data是我们命名的pydantic对象，model_dump()方法将pydantic对象转换成字典(键值对)→**解包# 上传信息没有设置值的空值会被忽略，只更新有值的字段query=update(User).where(User.username==username).values(**user_data.model_dump(exclude_unset=True,# 这两个设置是不接受前端没有传来的字段，以及不接受默认值为none的字段exclude_none=True))result=awaitdb.execute(query)awaitdb.commit()# 检查更新行数ifresult.rowcount==0:# 没有更新raiseHTTPException(status_code=422,detail="用户不存在")# 成功更新，通过用户名查询数据库update_user=awaitget_user(db,username)returnupdate_user

这里的user_data是一个pydantic类，model_dump是将其转化为字典对象，**是进一步将字典对象转化成关键字对象，可共update().values()使用。

四、用户修改密码接口

为了增强用户密码类型，特地在原项目基础上加上密码修改密码接口且密码加密贴合当今主流应用。

1.接口核心代码

@router.put("/password")asyncdefupdate_user_password(password_data:UserChangePasswordRequest,user:User=Depends(get_current_user),db:AsyncSession=Depends(get_db)):res_change_pwd=awaitusers.change_password(db,user,password_data.old_password,password_data.new_password)ifnotres_change_pwd:# 旧密码错误raiseHTTPException(status_code=400,detail="旧密码错误")returnsuccess_response(message="修改密码成功")

# 用正则表达加强密码classUserChangePasswordRequest(BaseModel):old_password:str=Field(...,alias="oldPassword",description="旧密码")new_password:str=Field(...,alias="newPassword",description="新密码(8-16位，且同时包含大写字母、小写字母和数字)")# 规则等价于 ^(?=.*[a-z])(?=.*[A-Z])(?=.*\d)[A-Za-z0-9!@#$%^&*.]{8,16}$@field_validator("new_password")# pydantic的类型装饰器，指定字段为python的new_password字段@classmethod# 类型defvalidate_new_password(cls,v:str):ifnotre.fullmatch(r"[A-Za-z0-9!@#$%^&*.]{8,16}",v):raiseValueError("新密码需8-16位，且仅允许包含字母、数字和 !@#$%^&*. 字符")ifnotre.search(r"[a-z]",v)ornotre.search(r"[A-Z]",v)ornotre.search(r"\d",v):raiseValueError("新密码必须同时包含大写字母、小写字母和数字")returnv# 修改密码的函数asyncdefchange_password(db:AsyncSession,user:User,old_password:str,new_password:str):ifnotverify_password(old_password,user.password):returnFalsehashed_new_pwd=get_hash_password(new_password)user.password=hashed_new_pwd db.add(user)# 这句是要理解的由sqlalchemy接管这个User对象,确保可以commit# 规避 session 是否过期或关闭导致不能提交的问题awaitdb.commit()awaitdb.refresh(user)returnTrue

这个接口是为了拓展正则表达式加强密码强度的学习，所以在修改密码的基础上加上了这个要求：正则表达式约束新密码。在实际开发中，其实在注册用户接口处对新密码的强度就是有所要求的。

五、用户头像上传接口

1.接口核心代码

这个接口是在项目基础上加一个修改头像的接口，实现用户可以自己上传喜欢的图片，极大丰富头条项目的用户信息板块。

@router.post("/avatar")asyncdefupdate_user_avatar(avatar_file:UploadFile=File(...),# 接受前端上传过来的图片。user:User=Depends(get_current_user),db:AsyncSession=Depends(get_db)):ext=ALLOWED_AVATAR_TYPES.get(avatar_file.content_type)# 前端http请求里面自带图片类型。ifnotext:raiseHTTPException(status_code=400,detail="仅支持 jpg/png/gif/webp 格式的头像")content=awaitavatar_file.read()iflen(content)>AVATAR_MAX_SIZE:raiseHTTPException(status_code=400,detail="头像图片大小不能超过2MB")# 文件名由服务端生成，避免使用用户上传的原始文件名(防路径穿越)filename=f"{user.id}_{uuid4().hex}.{ext}"(AVATAR_DIR/filename).write_bytes(content)avatar_path=f"/static/avatars/{filename}"# 这里是图片头像的存储策略，把图片和后端代码一起存在磁盘里面，数据库里面不存图片，直接存图片的访问路径。这样实现解耦。user.avatar=avatar_path db.add(user)awaitdb.commit()awaitdb.refresh(user)returnsuccess_response(message="头像更新成功",data={"avatar":avatar_path})# 这是在代码的主文件夹下创建一个名为：uploads的文件夹方便在磁盘里面存放图片。UPLOAD_DIR=Path(__file__).parent/"uploads"(UPLOAD_DIR/"avatars").mkdir(parents=True,exist_ok=True)app.mount("/static",StaticFiles(directory=str(UPLOAD_DIR)),name="static")