在求职平台项目中用户认证与文件上传是核心基础功能。今天我们将实现了机号密码登录、短信验证码登录、JWT Token 生成与解析以及头像上传到阿里云 OSS 的功能二、核心模块与技术选型认证方式密码登录 短信验证码登录Token 管理JWTaccess_token refresh_token验证码存储Redis设置过期时间防止重复获取文件存储阿里云 OSS存储用户头像框架分层接口层路由 服务层业务逻辑 模型层数据库三、数据库模型设计首先定义求职者Jobseeker模型存储用户核心信息python运行# app/models/jobseeker.py from tortoise import models, fields class Jobseeker(models.Model): id fields.IntField(pkTrue) phone fields.CharField(max_length12, description手机号, uniqueTrue) password fields.CharField(max_length12, description密码) nickname fields.CharField(max_length12, description昵称, default用户1) avatar fields.CharField(max_length255, description头像, default无) wxchat fields.CharField(max_length255, description微信号, nullTrue) email fields.CharField(max_length255, description邮箱, nullTrue) create_time fields.DatetimeField(auto_now_addTrue) update_time fields.DatetimeField(auto_nowTrue) class Meta: table jobseeker四、接口层实现路由我们使用APIRouter模块化管理求职者接口将路由前缀设为/jobseeker。1. 路由初始化python运行# app/api/jobseeker.py from fastapi import APIRouter, Depends, UploadFile, Request from app.schemas.jobseeker import JobseekerLoginPsd, JobseekerLoginSms from app.services.jobseeker_service import JobSeekerService from app.core.depends import getinfo_form_token jobseeker_router APIRouter(prefix/jobseeker, tags[求职者模块])2. 登录相关接口1手机号 密码登录python运行jobseeker_router.post(path/loginbypsd/, summary手机号密码登录) async def loginbypsd(logininfo: JobseekerLoginPsd): res await JobSeekerService.loginbypsd(logininfo) return {code: 200, msg: 登录成功, data: res}2发送短信验证码python运行jobseeker_router.get(path/getsms/, summary给指定手机号发送验证码) async def getsms(phone: str): res await JobSeekerService.getsms(phone) return {code: 200, msg: 发送成功}3手机号 验证码登录python运行jobseeker_router.post(path/loginbysms/, summary手机号验证码登录) async def loginbysms(logininfo: JobseekerLoginSms): res await JobSeekerService.loginbysms(logininfo) return {code: 200, msg: 登录成功, data: res}3. 登录后操作接口1获取个人信息通过Depends(getinfo_form_token)从 Token 中解析出用户 IDpython运行jobseeker_router.get(path/getmyinfo/, summary获取个人信息) async def getmyinfo(seekeridDepends(getinfo_form_token)): res await JobSeekerService.getmyinfo(seekerid) return {code: 200, msg: 获取成功, data: res}2忘记密码重置密码python运行jobseeker_router.put(path/getnewpsd/, summary忘记密码) async def getnewpsd(request: Request, seekeridDepends(getinfo_form_token)): newpsd await request.json() res await JobSeekerService.getnewpsd(newpsd.get(password), seekerid) return {code: 200, msg: 修改成功}3头像上传接收UploadFile类型文件上传到阿里云 OSSpython运行jobseeker_router.post(path/uploadavatar/, summary上传头像) async def uploadavatar(file: UploadFile, seekeridDepends(getinfo_form_token)): res await JobSeekerService.uploadavatar(file, seekerid) return {code: 200, msg: 上传成功}五、服务层实现业务逻辑1. Redis 连接初始化python运行import redis r redis.Redis(host127.0.0.1, port6379, db10, decode_responsesTrue, protocol2)2. 核心服务类JobSeekerService1手机号 密码登录python运行from app.models import Jobseeker from app.utils.jwt_util import create_tokens class JobSeekerService: staticmethod async def loginbypsd(logininfo: JobseekerLoginPsd): seekerobj await Jobseeker.get_or_none(phonelogininfo.phone) if seekerobj is None: raise Exception(手机号不存在) if seekerobj.password ! logininfo.password: raise Exception(密码错误) # 生成 JWT Token access_token, refresh_token create_tokens(seekerobj.id, seekerobj.phone) return { access_token: access_token, refresh_token: refresh_token, jobseeker: seekerobj }2发送短信验证码python运行import re from app.utils.sms_util import get_sms_code, send_sms_code from app.core.logging import logger staticmethod async def getsms(phone: str): if not phone: raise Exception(手机号不能为空) if not re.match(r^1[3-9]\d{9}$, phone): raise Exception(手机号格式不对) smskey fsms_code:{phone} # 防止重复获取验证码 code_redis r.get(smskey) if code_redis: raise Exception(请勿重复获取验证码) # 生成 6 位验证码 code str(get_sms_code(6)) # 调用第三方短信平台发送模拟 send_sms_code(phone, code) logger.info(f手机号为: {phone}验证码为: {code}) # 存入 Redis2 分钟过期 r.set(smskey, code, ex60*2) return True3手机号 验证码登录python运行staticmethod async def loginbysms(logininfo: JobseekerLoginSms): seekerobj await Jobseeker.get_or_none(phonelogininfo.phone) if seekerobj is None: raise Exception(用户不存在) smskey fsms_code:{logininfo.phone} code_redis r.get(smskey) if code_redis is None: raise Exception(验证码已过期请重新获取) if code_redis ! logininfo.code: raise Exception(验证码错误) # 生成 Token access_token, refresh_token create_tokens(seekerobj.id, seekerobj.phone) # 登录成功后删除验证码 r.delete(smskey) return { access_token: access_token, refresh_token: refresh_token, jobseeker: seekerobj }4获取用户信息python运行staticmethod async def getmyinfo(seekerid: str): seekerobj await Jobseeker.get_or_none(idseekerid) if not seekerobj: raise Exception(用户不存在) return seekerobj5重置密码python运行staticmethod async def getnewpsd(password: str, seekerid: str) - bool: seekerobj await Jobseeker.get_or_none(idseekerid) if not seekerobj: raise Exception(用户不存在) seekerobj.password password await seekerobj.save() return True6头像上传到阿里云 OSSpython运行from fastapi import UploadFile from app.utils.alioss_util import AliyunOSSTool staticmethod async def uploadavatar(file: UploadFile, seekerid: str): if not file: raise Exception(文件不能为空) # 读取文件内容 filedata await file.read() filename file.filename # 上传到阿里云 OSS alioss AliyunOSSTool() issuccess, result alioss.upload_single_file( filedata, filename, oss_pathp4boss/seeker_avatar/ ) if issuccess: seekerobj await Jobseeker.get_or_none(idseekerid) if not seekerobj: raise Exception(用户不存在) # 更新数据库头像地址 seekerobj.avatar result[access_url] await seekerobj.save() return True else: raise Exception(上传失败)六、关键技术点解析1. JWT Token 生成与解析生成登录成功后调用create_tokens()生成access_token短有效期和refresh_token长有效期。解析通过Depends(getinfo_form_token)中间件从请求头的Authorization中提取 Token解析出用户 ID实现接口鉴权。2. Redis 缓存验证码防重发获取验证码前先查询 Redis若存在则拒绝重复请求。过期控制验证码存入 Redis 时设置ex1202 分钟过期确保安全性。登录后销毁验证码登录成功后立即删除 Redis 中的验证码防止复用。3. 文件上传到阿里云 OSSUploadFileFastAPI 提供的文件上传类型可直接读取文件内容和文件名。OSS 工具类封装AliyunOSSTool实现文件上传、获取访问 URL 等功能。数据库更新上传成功后将 OSS 返回的access_url更新到用户avatar字段