FastAPI的基础了解和简单入门《五》求职者登录全套实战（双 Token JWT + Vue3 Element Plus 前端 + OSS 配置）

一、项目概述1. 需求分析求职者登录模块三种登录方式账号密码登录、手机号 短信验证码登录、钉钉第三方登录未绑定手机号则跳转绑定页配套能力用户注册、获取当前登录人信息接口认证方案业界标准双 Token 机制access_token短期有效配置 2 天用于访问所有需要登录的接口refresh_token长期有效配置 7 天专门用来无感刷新access_token避免频繁登录前后端交互规范前端请求受保护接口请求头携带Authorization: Bearer {access_token}access_token过期后前端使用refresh_token调用刷新接口获取全新双 Tokenrefresh_token失效则清空本地缓存强制跳转登录页配套资源阿里云 OSS 存储头像统一环境配置管理2. 数据库设计求职者表 t_job_seeker字段名类型约束说明idint主键、自增mobilevarchar唯一索引手机号登录凭证passwordvarcharbcrypt 加密存储不存明文avatarvarchar头像 OSS 访问链接nicknamevarchar用户昵称可为空emailvarchar邮箱可选wechatvarchar微信号可选create_timedatetime创建时间默认当前时间update_timedatetime更新时间修改自动更新二、核心流程详解1. 登录流程用户输入凭证用户在前端页面输入账号密码、手机号验证码或点击钉钉授权登录后端校验后端接收请求校验用户信息密码、验证码或第三方授权码生成双 Token校验通过后后端调用 JWT 工具类生成access_token短期和refresh_token长期返回前端将双 Token 和用户基本信息返回给前端前端存储前端将 Token 存储到本地如 localStorage 或 Vuex/Pinia并跳转至个人中心首页2. 鉴权请求流程发起请求前端访问需要登录的接口如/me获取个人信息携带 Token在请求头中携带Authorization: Bearer {access_token}后端校验后端 JWT 工具类校验 Token 的有效性、类型和过期时间解析用户校验通过后解析出用户 ID查询数据库获取用户完整信息返回数据将用户信息返回给前端完成鉴权3. Token 刷新流程Token 过期access_token过期后后端返回 401 鉴权失败状态码前端检测前端拦截器检测到 401 状态从本地读取refresh_token调用刷新接口前端使用refresh_token调用专门的 Token 刷新接口后端校验刷新后端校验refresh_token的有效性校验通过后生成全新的access_token返回新 Token将新的access_token返回给前端更新本地存储前端更新本地存储的access_token重试请求前端使用新的access_token重新发起之前失败的请求用户无感知完成刷新刷新失败如果refresh_token也失效则清空本地缓存强制跳转登录页4. 文件存储流程头像上传前端上传求职者在个人中心选择头像文件并上传后端直传 OSS后端接收文件后直接上传至阿里云 OSS获取访问链接OSS 返回文件的访问链接URL更新数据库后端将头像链接存入t_job_seeker.avatar字段前端展示前端个人中心读取该链接并展示用户头像三、后端核心开发FastAPI Tortoise-ORM JWT1. 依赖安装# 基础 web 框架 pip install fastapi uvicorn JWT 双 token 工具 pip install python-jose passlib bcrypt 数据库 ORM pip install tortoise-orm aiomysql 环境配置管理 pip install pydantic-settings 阿里云 OSS pip install oss22. 全局配置 config.pyPydantic 多环境配置from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): # 项目基础 app_env: str dev debug: bool True # JWT 配置 SECRET_KEY: str LTAI5t65xxxx阿里云 OSS 密钥 ALGORITHM: str HS256 ACCESS_TOKEN_EXPIRE_MINUTES: int 2880 # 2 天 REFRESH_TOKEN_EXPIRE_DAYS: int 7 # 7 天 阿里云 OSS 配置 OSS_ACCESS_KEY_ID: str LTAI5t65khxxxx OSS_ACCESS_KEY_SECRET: str l5dKxxxxxxxxx OSS_ENDPOINT: str oss-cn-beijing.aliyuncs.com OSS_BUCKET: str boss-job-file model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) settings Settings()3. JWT 工具类 app/utils/jwt_util.py双 Token 生成、校验、刷新from datetime import datetime, timedelta from typing import Optional, Tuple from jose import jwt, JWTError from app.config import settings def create_tokens(user_id: str, username: str) - Tuple[str, str]: 同时生成 access_token、refresh_token :param user_id: 用户 ID :param username: 手机号登录标识 :return: (access_token, refresh_token) # 公共载荷 common_payload { user_id: user_id, username: username } # 1. 短期访问 token access_payload common_payload.copy() access_payload[type] access access_expire datetime.utcnow() timedelta(minutessettings.ACCESS_TOKEN_EXPIRE_MINUTES) access_payload[exp] access_expire access_token jwt.encode(access_payload, settings.SECRET_KEY, algorithmsettings.ALGORITHM) 2. 长期刷新 token refresh_payload common_payload.copy() refresh_payload[type] refresh refresh_expire datetime.utcnow() timedelta(dayssettings.REFRESH_TOKEN_EXPIRE_DAYS) refresh_payload[exp] refresh_expire refresh_token jwt.encode(refresh_payload, settings.SECRET_KEY, algorithmsettings.ALGORITHM) return access_token, refresh_token def verify_token(token: str, token_type: str access) - Optional[dict]: 校验 token 合法性、类型、有效期 :param token: jwt 字符串 :param token_type: access / refresh :return: 载荷字典失败返回 None try: payload jwt.decode( token, settings.SECRET_KEY, algorithms[settings.ALGORITHM], options{verify_exp: True} ) 校验 token 类型 if payload.get(type) ! token_type: return None 校验必填字段 if not all(k in payload for k in [user_id, username]): return None return payload except JWTError: return None def refresh_access_token(refresh_token: str) - Optional[str]: 使用 refresh_token 刷新全新 access_token payload verify_token(refresh_token, token_typerefresh) if not payload: return None 生成新 access_token user_id payload[user_id] username payload[username] access_token, _ create_tokens(user_id, username) return access_token4. 阿里云 OSS 工具类 app/utils/oss_util.py头像上传、删除、权限控制import oss2 import uuid from datetime import datetime, timedelta from typing import Optional, Tuple from fastapi import HTTPException, UploadFile from app.config import settings class OSSUtil: 阿里云 OSS 文件上传、删除、访问权限控制工具类 def __init__(self): 初始化 OSS 客户端 auth oss2.Auth( settings.OSS_ACCESS_KEY_ID, settings.OSS_ACCESS_KEY_SECRET ) self.bucket oss2.Bucket( auth, settings.OSS_ENDPOINT, settings.OSS_BUCKET ) def generate_object_name(self, user_id: int, original_filename: str) -gt; str: 生成唯一的 OSS 对象名称 :param user_id: 用户 ID :param original_filename: 原始文件名 :return: 对象名称格式avatar/{user_id}/{timestamp_uuid}.{ext} # 提取文件扩展名 ext original_filename.split(.)[-1] if . in original_filename else jpg # 生成唯一文件名时间戳 UUID timestamp datetime.now().strftime(%Y%m%d%H%M%S) unique_id str(uuid.uuid4())[:8] # 构建对象路径 object_name favatar/{user_id}/{timestamp}_{unique_id}.{ext} return object_name async def upload_avatar(self, user_id: int, file: UploadFile) -gt; Tuple[str, str]: 上传用户头像到 OSS :param user_id: 用户 ID :param file: FastAPI UploadFile 对象 :return: (对象名称, 访问URL) try: # 1. 验证文件类型 allowed_types [image/jpeg, image/png, image/gif, image/webp] if file.content_type not in allowed_types: raise HTTPException( status_code400, detailf不支持的文件类型仅支持{, .join(allowed_types)} ) # 2. 验证文件大小限制 5MB max_size 5 * 1024 * 1024 # 5MB file_content await file.read() if len(file_content) amp;gt; max_size: raise HTTPException( status_code400, detail文件大小超过 5MB 限制 ) # 3. 生成唯一对象名称 object_name self.generate_object_name(user_id, file.filename) # 4. 上传到 OSS result self.bucket.put_object(object_name, file_content) if result.status ! 200: raise HTTPException( status_code500, detail文件上传到 OSS 失败 ) # 5. 生成访问 URL带签名30分钟有效 access_url self.generate_signed_url(object_name, expires1800) return object_name, access_url except Exception as e: if isinstance(e, HTTPException): raise e raise HTTPException( status_code500, detailf上传过程中发生错误{str(e)} ) def generate_signed_url(self, object_name: str, expires: int 1800) -gt; str: 生成带签名的临时访问 URL :param object_name: OSS 对象名称 :param expires: 过期时间秒默认30分钟 :return: 带签名的访问 URL try: # 生成带签名的 URL signed_url self.bucket.sign_url( GET, object_name, expires, slash_safeTrue ) return signed_url except Exception as e: raise HTTPException( status_code500, detailf生成签名 URL 失败{str(e)} ) def delete_avatar(self, object_name: str) -gt; bool: 删除 OSS 上的头像文件 :param object_name: OSS 对象名称 :return: 是否删除成功 try: # 检查文件是否存在 if not self.bucket.object_exists(object_name): return True # 文件不存在视为删除成功 # 删除文件 result self.bucket.delete_object(object_name) return result.status 204 except Exception as e: raise HTTPException( status_code500, detailf删除文件失败{str(e)} ) def set_object_acl(self, object_name: str, acl: str private) -gt; bool: 设置 OSS 对象的访问权限 :param object_name: OSS 对象名称 :param acl: 权限类型可选值private私有、public-read公共读、public-read-write公共读写 :return: 是否设置成功 try: # 验证权限类型 valid_acls [private, public-read, public-read-write] if acl not in valid_acls: raise ValueError(f无效的 ACL 类型可选值{, .join(valid_acls)}) # 设置对象 ACL self.bucket.put_object_acl(object_name, acl) return True except Exception as e: raise HTTPException( status_code500, detailf设置文件权限失败{str(e)} ) def get_object_info(self, object_name: str) -gt; Optional[dict]: 获取 OSS 对象信息 :param object_name: OSS 对象名称 :return: 对象信息字典包含大小、类型、最后修改时间等 try: # 获取对象元数据 meta self.bucket.get_object_meta(object_name) return { size: meta.content_length, type: meta.content_type, last_modified: meta.last_modified, etag: meta.etag, object_name: object_name } except oss2.exceptions.NoSuchKey: return None except Exception as e: raise HTTPException( status_code500, detailf获取文件信息失败{str(e)} ) 全局 OSS 工具实例 oss_util OSSUtil()5. 头像管理 Service整合 OSS 与数据库from fastapi import HTTPException, UploadFile from app.models.job_seeker import JobSeeker from app.utils.oss_util import oss_util import asyncio class AvatarService: 头像管理服务整合 OSS 操作与数据库更新 staticmethod async def upload_avatar(user_id: int, file: UploadFile) -gt; dict: 上传头像并更新数据库 :param user_id: 用户 ID :param file: 头像文件 :return: 上传结果信息 try: # 1. 查询用户信息 seeker await JobSeeker.get_or_none(iduser_id) if not seeker: raise HTTPException(status_code404, detail用户不存在) # 2. 如果用户已有头像先删除旧头像 old_avatar seeker.avatar if old_avatar: # 从 avatar URL 中提取 object_name # 假设 avatar 字段存储的是完整的 OSS 对象名称或 URL old_object_name old_avatar.split(/)[-1] if / in old_avatar else old_avatar try: oss_util.delete_avatar(favatar/{user_id}/{old_object_name}) except: # 删除旧头像失败不影响新头像上传记录日志即可 pass # 3. 上传新头像到 OSS object_name, access_url await oss_util.upload_avatar(user_id, file) # 4. 设置文件权限为私有默认 oss_util.set_object_acl(object_name, private) # 5. 更新数据库中的头像链接 # 存储完整的带签名 URL30分钟有效 seeker.avatar access_url await seeker.save() # 6. 返回上传结果 return { success: True, message: 头像上传成功, avatar_url: access_url, object_name: object_name, expires_in: 1800 # 30分钟 } except HTTPException: raise except Exception as e: raise HTTPException( status_code500, detailf头像上传失败{str(e)} ) staticmethod async def get_avatar_url(user_id: int) -gt; Optional[str]: 获取用户头像访问 URL带签名 :param user_id: 用户 ID :return: 带签名的头像 URL如果用户没有头像则返回 None try: # 查询用户信息 seeker await JobSeeker.get_or_none(iduser_id) if not seeker or not seeker.avatar: return None # 从数据库中的 avatar 字段提取 object_name # 这里假设 avatar 字段存储的是完整的 URL需要提取 object_name avatar_url seeker.avatar if avatar/ in avatar_url: # 从 URL 中提取 object_name parts avatar_url.split(avatar/) if len(parts) amp;gt; 1: object_name favatar/{parts[1].split(?)[0]} else: object_name seeker.avatar else: object_name seeker.avatar # 生成新的带签名 URL30分钟有效 signed_url oss_util.generate_signed_url(object_name, expires1800) # 更新数据库中的 URL可选可以定期清理过期 URL seeker.avatar signed_url await seeker.save() return signed_url except Exception as e: # 如果生成签名 URL 失败返回原始 URL return seeker.avatar if seeker and seeker.avatar else None staticmethod async def delete_avatar(user_id: int) -gt; dict: 删除用户头像 :param user_id: 用户 ID :return: 删除结果信息 try: # 查询用户信息 seeker await JobSeeker.get_or_none(iduser_id) if not seeker: raise HTTPException(status_code404, detail用户不存在) if not seeker.avatar: return { success: True, message: 用户没有头像无需删除 } # 从 avatar URL 中提取 object_name avatar_url seeker.avatar if avatar/ in avatar_url: parts avatar_url.split(avatar/) if len(parts) amp;gt; 1: object_name favatar/{parts[1].split(?)[0]} else: object_name seeker.avatar else: object_name seeker.avatar # 删除 OSS 上的文件 delete_result oss_util.delete_avatar(object_name) if delete_result: # 清空数据库中的头像链接 seeker.avatar None await seeker.save() return { success: True, message: 头像删除成功 } else: return { success: False, message: 头像删除失败 } except HTTPException: raise except Exception as e: raise HTTPException( status_code500, detailf头像删除失败{str(e)} ) staticmethod async def refresh_avatar_url(user_id: int) -gt; Optional[str]: 刷新用户头像 URL 签名防止 URL 过期 :param user_id: 用户 ID :return: 新的带签名 URL return await AvatarService.get_avatar_url(user_id)/code/pre4. 求职者业务 Service登录、获取个人信息from fastapi import HTTPException from app.models.job_seeker import JobSeeker from app.utils.jwt_util import create_tokens from passlib.context import CryptContext pwd_context CryptContext(schemes[bcrypt], deprecatedauto) class JobSeekerService: staticmethod async def login_by_mobile_pwd(mobile: str, password: str): 手机号密码登录返回用户信息双 token # 查询用户 seeker await JobSeeker.get_or_none(mobilemobile) if not seeker or not pwd_context.verify(password, seeker.password): raise HTTPException(status_code401, detail手机号或密码错误) # 生成双 token access_token, refresh_token create_tokens(str(seeker.id), seeker.mobile) return { jobseeker: seeker, access_token: access_token, refresh_token: refresh_token } staticmethod async def get_current_seeker(authorization: str): 解析 header token获取当前登录求职者 if not authorization or not authorization.startswith(Bearer ): raise HTTPException(status_code401, detail认证信息无效) token authorization.split( )[1] payload verify_token(token, token_typeaccess) if not payload: raise HTTPException(status_code401, detailToken 无效或已过期) seeker await JobSeeker.get_or_none(idint(payload[user_id])) if not seeker: raise HTTPException(status_code401, detail用户不存在) return seeker/code/pre