基于 FastAPI Tortoise ORM Redis JWT实现求职者端与企业端双角色登录体系。 代码分块讲解可直接跟着实现。目录整体架构设计项目结构搭建数据模型定义验证码服务 — Redis 方案JWT 工具封装请求模型 Schema核心登录服务API 路由层全局异常处理前端对接要点安全加固建议1. 整体架构设计多端登录的核心问题是同一个后端服务为不同角色求职者、企业提供独立的登录入口、token 体系与权限隔离。设计原则原则说明入口隔离求职者/job_seeker/login企业/enterprise/login路径分明Token 区分求职者 token 与企业 token 使用不同的 key 前缀互不通用验证码复用一个发送接口 scene 参数按场景分发到不同 Redis key权限守卫后续中间件根据 token 类型拦截跨端访问数据流向图┌─────────────┐ POST /send_sms_code/{mobile}?scenelogin ┌──────────┐ │ 前端(求职者) │ ──────────────────────────────────────────────→ │ Redis │ │ │ boss-api:login:sms:{mobile} 649944 │ (db6) │ │ │ ←────────────────────────────────────────────── │ │ │ │ POST /job_seeker/login_by_mobile_code │ │ │ │ → 校验 Redis → 生成 JWT → 返回 token │ │ └─────────────┘ └──────────┘ ┌─────────────┐ POST /send_sms_code/{mobile}?sceneenterprise-login ┌──────────┐ │ 前端(企业) │ ────────────────────────────────────────────────────→ │ Redis │ │ │ boss-api:enterprise-login:sms:{mobile} 325724 │ (db6) │ │ │ ←──────────────────────────────────────────────────── │ │ │ │ POST /enterprise/login │ │ │ │ → 校验 Redis → 生成 JWT → 返回 token │ │ └─────────────┘ └──────────┘2. 项目结构搭建参考上一篇分层架构3. 数据模型定义求职者模型python复制# app/models/job_seeker.py from tortoise import fields, models class JobSeeker(models.Model): 求职者 — 登录的主表 id fields.IntField(pkTrue) mobile fields.CharField(max_length20, uniqueTrue, description手机号) password fields.CharField(max_length100, nullTrue, description密码(哈希)) nickname fields.CharField(max_length50, nullTrue, description昵称) avatar fields.CharField(max_length200, nullTrue, description头像URL) created_at fields.DatetimeField(auto_now_addTrue) class Meta: table job_seeker企业相关模型企业登录校验的是企业资质表中的联系手机号而非企业主表本身。python复制# app/models/enterprise.py from tortoise import fields, models from enum import IntEnum class AccountStatus(IntEnum): PENDING_AUDIT 0 # 待审核 NORMAL 1 # 正常审核通过 DISABLED 2 # 已封禁 class BlackListStatus(IntEnum): NOT_BANNED 0 BANNED 1 class AuditType(IntEnum): NEW_ENTERPRISE_AUTH 0 # 新企业认证 UPDATE_AUTH 1 # 更新认证 class Enterprise(models.Model): 企业主表 id fields.IntField(pkTrue) enterprise_name fields.CharField(max_length100, description企业名称) enterprise_code fields.CharField(max_length50, uniqueTrue, description企业编码) account_status fields.IntEnumField( enum_typeAccountStatus, defaultAccountStatus.PENDING_AUDIT, description账号状态 ) blacklist_status fields.IntEnumField( enum_typeBlackListStatus, defaultBlackListStatus.NOT_BANNED, description黑名单状态 ) submit_time fields.DatetimeField(nullTrue, description提交时间) class Meta: table enterprise class EnterpriseQualification(models.Model): 企业资质 — 含联系手机号是登录校验的关键表 id fields.IntField(pkTrue) contact_name fields.CharField(max_length50, description联系人姓名) contact_phone fields.CharField(max_length20, description联系手机号 ← 登录用此字段) contact_email fields.CharField(max_length100, nullTrue, description联系邮箱) enterprise_id fields.IntField(description关联企业ID) business_license_url fields.CharField(max_length200, nullTrue) legal_id_front_url fields.CharField(max_length200, nullTrue) legal_id_back_url fields.CharField(max_length200, nullTrue) class Meta: table enterprise_qualification class EnterpriseReview(models.Model): 企业审核记录 — 登录时检查是否审核通过 id fields.IntField(pkTrue) enterprise_id fields.IntField(description关联企业ID) review_result fields.IntField(description审核结果: 1通过, 0未通过) review_reason fields.CharField(max_length200, nullTrue, description审核原因) remark fields.CharField(max_length200, nullTrue, description备注) review_time fields.DatetimeField(nullTrue, description审核时间) class Meta: table enterprise_review关键设计点企业登录通过EnterpriseQualification.contact_phone查找用户再通过EnterpriseReview.review_result 1确认审核通过。两层校验保证了未审核企业无法登录。4. 验证码服务 — Redis 方案这是多端登录的核心基础设施。一个发送接口通过scene参数将验证码分发到不同的 Redis key。Redis 连接python复制# 在 service 文件顶部初始化也可抽为独立模块 import redis redis_client redis.Redis( hostlocalhost, port6379, db6, # 独立 db避免与其他业务冲突 decode_responsesTrue, protocol2 )验证码生成与存储# app/services/job_seeker.py节选 from app.utils.generate_code_util import generate_code from app.utils.sms import RongLianSms # 容联云短信 SDK可替换 class JobSeekerService: staticmethod async def send_sms_code(mobile: str, scene: str): 发送验证码 — 多场景复用 Redis key 格式: boss-api:{scene}:sms:{mobile} scene 取值: - login → boss-api:login:sms:{mobile} 求职者登录 - register → boss-api:register:sms:{mobile} 求职者注册 - enterprise-login → boss-api:enterprise-login:sms:{mobile} 企业登录 这就是多端登录的关键同一个发送接口不同 scene 落入不同 Redis key 各端登录时只校验自己场景的 key互不干扰。 # ── 第1步注册场景特殊校验 ── if scene register: jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is not None: raise Exception(该手机号已注册请直接登录) # 登录场景不做此校验因为企业手机号不在 JobSeeker 表中 # ── 第2步生成6位随机验证码 ── code generate_code(6) # 例: 325724 # ── 第3步存入 Redis2分钟过期 ── key fboss-api:{scene}:sms:{mobile} redis_client.set(key, code, ex120) # ex120秒 # ── 第4步发送短信 ── # 实际项目中调用短信 SDK # sms RongLianSms() # sms.send_sms(mobile, code) # 开发阶段可跳过发送直接在日志中打印验证码 return {message: 验证码发送成功}复制Redis Key 对照表场景 Redis Key 过期时间 ───────────────────────────────────────────────────────────────── 求职者登录 boss-api:login:sms:{mobile} 120s 求职者注册 boss-api:register:sms:{mobile} 120s 企业登录 boss-api:enterprise-login:sms:{mobile} 120s 手机号绑定 boss-api:bind:sms:{mobile} 120s 可继续扩展...为什么不用单独的企业验证码接口因为验证码的「发送」逻辑完全相同生成随机数 → 存 Redis → 发短信唯一的区别就是 Redis key 前缀。用 scene 参数分发是最高效的设计新增场景只需传一个新 scene 值无需写任何新代码。5. JWT 工具封装多端登录需要生成带角色标识的 token以便后续中间件区分 token 类型。# app/utils/jwt_util.py import jwt from datetime import datetime, timedelta JWT_SECRET your-secret-key # 生产环境应从环境变量读取 JWT_ALGORITHM HS256 # 两端使用不同过期策略 ACCESS_TOKEN_EXPIRE_MINUTES 30 # access token 30分钟 REFRESH_TOKEN_EXPIRE_DAYS 7 # refresh token 7天 def create_tokens(user_id: str, mobile: str, role: str job_seeker): 生成 JWT token 对access refresh role 取值: - job_seeker → 求职者 - enterprise → 企业端 token payload 结构: { sub: user_id, # 用户/企业ID mobile: mobile, # 手机号 role: role, # ← 角色标识中间件据此拦截跨端访问 type: access, # token 类型access / refresh exp: 过期时间戳 } now datetime.utcnow() # Access Token — 短期有效 access_payload { sub: user_id, mobile: mobile, role: role, type: access, exp: now timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES), } access_token jwt.encode(access_payload, JWT_SECRET, algorithmJWT_ALGORITHM) # Refresh Token — 长期有效仅用于换取新 access token refresh_payload { sub: user_id, mobile: mobile, role: role, type: refresh, exp: now timedelta(daysREFRESH_TOKEN_EXPIRE_DAYS), } refresh_token jwt.encode(refresh_payload, JWT_SECRET, algorithmJWT_ALGORITHM) return access_token, refresh_token def decode_token(token: str): 解析 JWT token返回 payload dict 过期或无效时会抛出 jwt.ExpiredSignatureError / jwt.InvalidTokenError return jwt.decode(token, JWT_SECRET, algorithms[JWT_ALGORITHM]) def verify_access_token(token: str, expected_role: str None): 校验 access token可选校验角色 1. 解析 token 2. 确认 type access不接受 refresh token 直接访问 3. 如指定 expected_role确认 role 匹配 payload decode_token(token) if payload.get(type) ! access: raise Exception(无效的 token 类型需要 access token) if expected_role and payload.get(role) ! expected_role: raise Exception(f权限不足需要 {expected_role} 角色) return payload复制Token Payload 示例json复制// 求职者 access token { sub: 42, mobile: 15673057586, role: job_seeker, type: access, exp: 1722195120 } // 企业端 access token { sub: 7, mobile: 15673057586, role: enterprise, type: access, exp: 1722195120 }角色字段role是多端隔离的关键后续所有受保护接口都应通过中间件校验role确保求职者 token 不能访问企业端接口反之亦然。6. 请求模型 Schema# app/schemas/job_seeker.py from pydantic import BaseModel, Field class LoginByMobileCodeRequest(BaseModel): 求职者 — 手机号 验证码登录 mobile: str Field(..., description手机号) code: str Field(..., description短信验证码) class RegisterRequest(BaseModel): 求职者 — 注册验证码 密码 mobile: str Field(..., description手机号) code: str Field(..., description短信验证码) password: str Field(..., min_length6, description密码) # app/schemas/enterprise.py from pydantic import BaseModel, Field class LoginMobileRequest(BaseModel): 企业端 — 手机号 验证码登录 mobile: str Field(..., description联系手机号) code: str Field(..., description验证码)复制7. 核心登录服务求职者登录# app/services/job_seeker.py登录部分 class JobSeekerService: staticmethod async def login_by_mobile_code(mobile: str, code: str): 求职者验证码登录流程: 1. 查找用户手机号必须在 JobSeeker 表中存在 2. 校验 Redis 中的验证码key: boss-api:login:sms:{mobile} 3. 生成 JWT token 4. 删除已使用的验证码 # ── 第1步查找用户 ── jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(该手机号未注册) # ── 第2步校验验证码 ── key fboss-api:login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! code: raise Exception(验证码错误) # ── 第3步生成 tokenrolejob_seeker── access_token, refresh_token create_tokens( str(jobseeker.id), mobile, rolejob_seeker ) # ── 第4步删除验证码防止复用 ── redis_client.delete(key) return { access_token: access_token, refresh_token: refresh_token, user_info: { id: jobseeker.id, mobile: jobseeker.mobile, nickname: jobseeker.nickname, } }复制企业登录# app/services/enterprise_service.py登录部分 class EnterpriseService: staticmethod async def login(loginMobileRequest: LoginMobileRequest): 企业端验证码登录流程: 1. 通过联系手机号查找企业资质记录 2. 检查对应企业是否审核通过 3. 校验 Redis 中的验证码key: boss-api:enterprise-login:sms:{mobile} 4. 生成 JWT tokenroleenterprise 5. 删除已使用的验证码 注意企业登录校验的是 EnterpriseQualification.contact_phone 字段 而不是 Enterprise 主表本身。一个手机号可能关联多家企业HR可管理多个。 mobile loginMobileRequest.mobile # ── 第1步通过联系手机号查找资质记录 ── enterprise_qualifications await EnterpriseQualification.filter( contact_phonemobile ) if not enterprise_qualifications: raise Exception(手机号不存在) # ── 第2步逐一检查关联企业的审核状态 ── for enterprise_qualification in enterprise_qualifications: enterprise_id enterprise_qualification.enterprise_id # 获取审核记录 enterprise_review await EnterpriseReview.get_or_none( enterprise_identerprise_id ) # 审核通过才继续 if enterprise_review and enterprise_review.review_result 1: # ── 第3步校验验证码 ── # 关键key 使用 enterprise-login 场景前缀 key fboss-api:enterprise-login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! loginMobileRequest.code: raise Exception(验证码错误) # ── 第4步生成 tokenroleenterprise── access_token, refresh_token create_tokens( str(enterprise_id), mobile, roleenterprise ) # ── 第5步删除验证码 ── redis_client.delete(key) return { enterprise_access_token: access_token, enterprise_refresh_token: refresh_token, } # 所有关联企业都未审核通过 raise Exception(账号未审核通过)复制两端登录服务的差异对照维度求职者企业用户查找JobSeeker.mobileEnterpriseQualification.contact_phone额外校验无注册即可登录需审核通过review_result 1Redis keyboss-api:login:sms:{mobile}boss-api:enterprise-login:sms:{mobile}JWT rolejob_seekerenterpriseToken 字段名access_tokenenterprise_access_token8. API 路由层求职者路由# app/apis/job_seeker_api.py from fastapi import APIRouter, Query from app.schemas.job_seeker import LoginByMobileCodeRequest from app.services.job_seeker import JobSeekerService job_seeker_router APIRouter(prefix/job_seeker, tags[求职者]) job_seeker_router.post( /send_sms_code/{mobile}, summary发送验证码, description按场景发送验证码scene 参数决定 Redis key 前缀 ) async def send_sms_code( mobile: str, scene: str Query(login, title场景login/register/enterprise-login/bind) ): 验证码发送 — 多端共用 scene 参数说明: - login: 求职者登录验证码 - register: 求职者注册验证码 - enterprise-login: 企业登录验证码 ← 企业前端传此值 - bind: 绑定手机号验证码 同一个接口服务所有场景无需为每个端写单独的发送接口。 res await JobSeekerService.send_sms_code(mobile, scene) return {code: 1, message: 验证码发送成功, data: res} job_seeker_router.post( /login_by_mobile_code, summary手机号验证码登录, description求职者登录 ) async def login_by_mobile_code(request: LoginByMobileCodeRequest): res await JobSeekerService.login_by_mobile_code(request.mobile, request.code) return {code: 1, message: 登录成功, data: res}复制企业路由复制# app/apis/enterprise_api.py from fastapi import APIRouter from app.schemas.enterprise import LoginMobileRequest from app.services.enterprise_service import EnterpriseService enterprise_router APIRouter(prefix/enterprise, tags[企业端]) enterprise_router.post( /login, summary企业登录, description企业端通过联系手机号 验证码登录 ) async def enterprise_login(request: LoginMobileRequest): res await EnterpriseService.login(request) return {code: 1, message: 登录成功, data: res}路由挂载# app/main.py from fastapi import FastAPI from app.core.middlewares import add_cors_middleware from app.core.exception_handler import global_exception_handler from app.apis.job_seeker_api import job_seeker_router from app.apis.enterprise_api import enterprise_router from app.database import init_db app FastAPI(titleBoss API, version1.0.0) # 中间件 add_cors_middleware(app) # 全局异常处理 app.add_exception_handler(Exception, global_exception_handler) # 挂载路由 app.include_router(job_seeker_router) app.include_router(enterprise_router) # 数据库初始化事件 app.on_event(startup) async def startup(): await init_db()复制9. 全局异常处理统一的异常处理确保所有raise Exception(xxx)都返回格式一致的 JSON。# app/core/exception_handler.py from starlette.requests import Request from starlette.responses import JSONResponse from app.core.logging import logger def global_exception_handler(request: Request, exc: Exception) - JSONResponse: 全局异常处理器 所有业务层 raise Exception(错误信息) 都会被这里捕获 统一返回 HTTP 500 {code: 0, message: 错误信息} 前端只需: - code 1 → 成功 - code 0 → 失败message 中有具体原因 logger.error(f异常: {str(exc)}) return JSONResponse( status_code500, content{ code: 0, message: str(exc) } )复制响应格式统一标准成功响应 (HTTP 200): { code: 1, message: 登录成功, data: { ... } } 失败响应 (HTTP 500): { code: 0, message: 验证码已过期 ← 来自业务层 raise Exception() }10. 前端对接要点Axios 响应拦截器适配后端code: 1表示成功code: 0表示失败HTTP 500。前端需要调整拦截器javascript// src/utils/request.js service.interceptors.response.use( (response) { const res response.data // 后端 code1 表示成功 if (res.code 1) { return res // 直接返回业务代码用 res.data 取数据 } // code0 或其他 → 业务错误 ElMessage.error(res.message || 请求失败) return Promise.reject(new Error(res.message)) }, (error) { // HTTP 500 来自后端全局异常处理 if (error.response?.status 500) { const data error.response.data ElMessage.error(data?.message || 服务器内部错误) } return Promise.reject(error) } )复制验证码发送企业端javascript复制// src/api/login.js import request from /utils/request export function sendSmsCode(mobile) { // 关键sceneenterprise-login 让验证码存入企业登录专属的 Redis key return request.post(/job_seeker/send_sms_code/${mobile}, null, { params: { scene: enterprise-login } }) } export function enterpriseLogin(mobile, code) { return request.post(/enterprise/login, { mobile, code }) }验证码发送求职者端javascript复制// 求职者端 scenelogin默认值可不传 export function sendSeekerSmsCode(mobile) { return request.post(/job_seeker/send_sms_code/${mobile}, null, { params: { scene: login } }) } export function seekerLogin(mobile, code) { return request.post(/job_seeker/login_by_mobile_code, { mobile, code }) }Token 存储两端分离javascript复制// 求职者端 localStorage.setItem(seekerToken, res.data.access_token) localStorage.setItem(seekerRefreshToken, res.data.refresh_token) // 企业端 localStorage.setItem(companyToken, res.data.enterprise_access_token) localStorage.setItem(companyRefreshToken, res.data.enterprise_refresh_token)两端使用不同的 localStorage key即使同一浏览器打开两个端也不会互相干扰。11. 安全加固建议已实现的安全措施措施说明验证码过期Redisex1202分钟自动失效一次性使用登录成功后redis_client.delete(key)删除验证码注册场景校验sceneregister时检查手机号是否已注册企业审核校验企业登录需review_result 1才能通过建议补充的措施(1) 验证码发送频率限制python复制# 在 send_sms_code 方法开头添加 send_key fboss-api:sms-limit:{mobile} count redis_client.get(send_key) if count and int(count) 5: raise Exception(发送次数过多请稍后再试) redis_client.incr(send_key) redis_client.expire(send_key, 3600) # 1小时内最多5次(2) 登录失败次数限制python复制# 在 login 方法开头添加 fail_key fboss-api:login-fail:{mobile} fail_count redis_client.get(fail_key) if fail_count and int(fail_count) 5: raise Exception(登录失败次数过多请30分钟后重试)python复制# 登录失败时 redis_client.incr(fail_key) redis_client.expire(fail_key, 1800) # 30分钟 # 登录成功时 redis_client.delete(fail_key)(3) Token 角色守卫中间件# app/core/auth_guard.py from fastapi import Request, HTTPException from starlette.middleware.base import BaseHTTPMiddleware from app.utils.jwt_util import verify_access_token class RoleGuardMiddleware(BaseHTTPMiddleware): 根据 URL 前缀自动校验 token 角色 /enterprise/* → 需要 roleenterprise /job_seeker/* → 需要 rolejob_seeker 公开接口如 send_sms_code 不受此守卫影响 ROLE_MAP { /enterprise: enterprise, /job_seeker: job_seeker, } # 不需要 token 的路径 WHITE_LIST [ /job_seeker/send_sms_code, /job_seeker/login_by_mobile_code, /job_seeker/register, /enterprise/login, ] async def dispatch(self, request: Request, call_next): path request.url.path # 白名单直接放行 if any(path.startswith(w) for w in self.WHITE_LIST): return await call_next(request) # 查找对应的角色要求 required_role None for prefix, role in self.ROLE_MAP.items(): if path.startswith(prefix): required_role role break if required_role is None: return await call_next(request) # 提取 token auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detail未提供 token) token auth_header[7:] try: payload verify_access_token(token, expected_rolerequired_role) # 可将 payload 存入 request.state 供后续使用 request.state.user payload except Exception as e: raise HTTPException(status_code401, detailstr(e)) return await call_next(request)复制(4) Refresh Token 换新 Access Token# app/apis/token_api.py from fastapi import APIRouter from app.utils.jwt_util import create_tokens, decode_token token_router APIRouter(prefix/token, tags[Token管理]) token_router.post(/refresh, summary刷新 Access Token) async def refresh_token(refresh_token: str): 使用 refresh token 换取新的 access token 流程: 1. 解析 refresh token 2. 确认 type refresh 3. 用其中的 sub/mobile/role 生成新 access token 4. 返回新 tokenrefresh token 本身不更换直到它也过期 payload decode_token(refresh_token) if payload.get(type) ! refresh: raise Exception(无效的 token 类型) new_access, _ create_tokens( payload[sub], payload[mobile], payload[role] ) return {code: 1, data: {access_token: new_access}}复制总结多端登录的本质不是写多个登录接口而是在共享的基础设施验证码服务、JWT工具上通过参数化scene、role实现场景分发与权限隔离验证码分发scene参数 → 不同 Redis key → 各端校验各自的 keyToken 隔离JWTrole字段 → 中间件按 role 拦截跨端请求业务差异各端 login service 内部逻辑不同用户查找表、额外校验条件但调用方式一致这套架构天然支持扩展——新增「HR 端」「管理员端」只需新增一个 scene 值、新增一个 login service、新增一个路由验证码发送接口零修改。