FastAPI 多端登录实现指南

基于 FastAPI + Tortoise ORM + Redis + JWT，实现求职者端与企业端双角色登录体系。 代码分块讲解，可直接跟着实现。

---

目录

1. 整体架构设计
2. 项目结构搭建
3. 数据模型定义
4. 验证码服务 — Redis 方案
5. JWT 工具封装
6. 请求模型 Schema
7. 核心登录服务
8. API 路由层
9. 全局异常处理
10. 前端对接要点
11. 安全加固建议

---

1. 整体架构设计

多端登录的核心问题是：同一个后端服务，为不同角色（求职者、企业）提供独立的登录入口、token 体系与权限隔离。

设计原则

原则	说明
入口隔离	求职者/job_seeker/login，企业/enterprise/login，路径分明
Token 区分	求职者 token 与企业 token 使用不同的 key 前缀，互不通用
验证码复用	一个发送接口 + scene 参数，按场景分发到不同 Redis key
权限守卫	后续中间件根据 token 类型拦截跨端访问

数据流向图

┌─────────────┐ POST /send_sms_code/{mobile}?scene=login ┌──────────┐ │ 前端(求职者) │ ──────────────────────────────────────────────→ │ Redis │ │ │ boss-api:login:sms:{mobile} = "649944" │ (db=6) │ │ │ ←────────────────────────────────────────────── │ │ │ │ POST /job_seeker/login_by_mobile_code │ │ │ │ → 校验 Redis → 生成 JWT → 返回 token │ │ └─────────────┘ └──────────┘ ┌─────────────┐ POST /send_sms_code/{mobile}?scene=enterprise-login ┌──────────┐ │ 前端(企业) │ ────────────────────────────────────────────────────→ │ Redis │ │ │ boss-api:enterprise-login:sms:{mobile} = "325724" │ (db=6) │ │ │ ←──────────────────────────────────────────────────── │ │ │ │ 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(pk=True) mobile = fields.CharField(max_length=20, unique=True, description="手机号") password = fields.CharField(max_length=100, null=True, description="密码(哈希)") nickname = fields.CharField(max_length=50, null=True, description="昵称") avatar = fields.CharField(max_length=200, null=True, description="头像URL") created_at = fields.DatetimeField(auto_now_add=True) 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(pk=True) enterprise_name = fields.CharField(max_length=100, description="企业名称") enterprise_code = fields.CharField(max_length=50, unique=True, description="企业编码") account_status = fields.IntEnumField( enum_type=AccountStatus, default=AccountStatus.PENDING_AUDIT, description="账号状态" ) blacklist_status = fields.IntEnumField( enum_type=BlackListStatus, default=BlackListStatus.NOT_BANNED, description="黑名单状态" ) submit_time = fields.DatetimeField(null=True, description="提交时间") class Meta: table = "enterprise" class EnterpriseQualification(models.Model): """企业资质 — 含联系手机号，是登录校验的关键表""" id = fields.IntField(pk=True) contact_name = fields.CharField(max_length=50, description="联系人姓名") contact_phone = fields.CharField(max_length=20, description="联系手机号 ← 登录用此字段") contact_email = fields.CharField(max_length=100, null=True, description="联系邮箱") enterprise_id = fields.IntField(description="关联企业ID") business_license_url = fields.CharField(max_length=200, null=True) legal_id_front_url = fields.CharField(max_length=200, null=True) legal_id_back_url = fields.CharField(max_length=200, null=True) class Meta: table = "enterprise_qualification" class EnterpriseReview(models.Model): """企业审核记录 — 登录时检查是否审核通过""" id = fields.IntField(pk=True) enterprise_id = fields.IntField(description="关联企业ID") review_result = fields.IntField(description="审核结果: 1=通过, 0=未通过") review_reason = fields.CharField(max_length=200, null=True, description="审核原因") remark = fields.CharField(max_length=200, null=True, description="备注") review_time = fields.DatetimeField(null=True, 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( host="localhost", port=6379, db=6, # 独立 db，避免与其他业务冲突 decode_responses=True, protocol=2 )

验证码生成与存储

# 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(mobile=mobile) if jobseeker is not None: raise Exception("该手机号已注册，请直接登录") # 登录场景不做此校验，因为企业手机号不在 JobSeeker 表中 # ── 第2步：生成6位随机验证码 ── code = generate_code(6) # 例: "325724" # ── 第3步：存入 Redis，2分钟过期 ── key = f"boss-api:{scene}:sms:{mobile}" redis_client.set(key, code, ex=120) # ex=120秒 # ── 第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(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), } access_token = jwt.encode(access_payload, JWT_SECRET, algorithm=JWT_ALGORITHM) # Refresh Token — 长期有效，仅用于换取新 access token refresh_payload = { "sub": user_id, "mobile": mobile, "role": role, "type": "refresh", "exp": now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS), } refresh_token = jwt.encode(refresh_payload, JWT_SECRET, algorithm=JWT_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_length=6, 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(mobile=mobile) if jobseeker is None: raise Exception("该手机号未注册") # ── 第2步：校验验证码 ── key = f"boss-api:login:sms:{mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != code: raise Exception("验证码错误") # ── 第3步：生成 token（role="job_seeker"）── access_token, refresh_token = create_tokens( str(jobseeker.id), mobile, role="job_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 token（role="enterprise"） 5. 删除已使用的验证码 注意：企业登录校验的是 EnterpriseQualification.contact_phone 字段， 而不是 Enterprise 主表本身。一个手机号可能关联多家企业（HR可管理多个）。 """ mobile = loginMobileRequest.mobile # ── 第1步：通过联系手机号查找资质记录 ── enterprise_qualifications = await EnterpriseQualification.filter( contact_phone=mobile ) 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_id=enterprise_id ) # 审核通过才继续 if enterprise_review and enterprise_review.review_result == 1: # ── 第3步：校验验证码 ── # 关键：key 使用 "enterprise-login" 场景前缀 key = f"boss-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步：生成 token（role="enterprise"）── access_token, refresh_token = create_tokens( str(enterprise_id), mobile, role="enterprise" ) # ── 第5步：删除验证码 ── redis_client.delete(key) return { "enterprise_access_token": access_token, "enterprise_refresh_token": refresh_token, } # 所有关联企业都未审核通过 raise Exception("账号未审核通过")

复制

两端登录服务的差异对照：

维度	求职者	企业
用户查找	JobSeeker.mobile	EnterpriseQualification.contact_phone
额外校验	无（注册即可登录）	需审核通过（review_result == 1）
Redis key	boss-api:login:sms:{mobile}	boss-api:enterprise-login:sms:{mobile}
JWT role	"job_seeker"	"enterprise"
Token 字段名	access_token	enterprise_access_token

---

8. 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(title="Boss API", version="1.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_code=500, 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 // 后端 code=1 表示成功 if (res.code === 1) { return res // 直接返回，业务代码用 res.data 取数据 } // code=0 或其他 → 业务错误 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) { // 关键：scene=enterprise-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

复制

// 求职者端 scene=login（默认值，可不传） 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. 安全加固建议

已实现的安全措施

措施	说明
验证码过期	Redisex=120，2分钟自动失效
一次性使用	登录成功后redis_client.delete(key)删除验证码
注册场景校验	scene=register时检查手机号是否已注册
企业审核校验	企业登录需review_result == 1才能通过

建议补充的措施

(1) 验证码发送频率限制

python

复制

# 在 send_sms_code 方法开头添加 send_key = f"boss-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 = f"boss-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/* → 需要 role="enterprise" /job_seeker/* → 需要 role="job_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_code=401, detail="未提供 token") token = auth_header[7:] try: payload = verify_access_token(token, expected_role=required_role) # 可将 payload 存入 request.state 供后续使用 request.state.user = payload except Exception as e: raise HTTPException(status_code=401, detail=str(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. 返回新 token（refresh 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 → 各端校验各自的 key
• Token 隔离：JWTrole字段 → 中间件按 role 拦截跨端请求
• 业务差异：各端 login service 内部逻辑不同（用户查找表、额外校验条件），但调用方式一致

这套架构天然支持扩展——新增「HR 端」「管理员端」只需：新增一个 scene 值、新增一个 login service、新增一个路由，验证码发送接口零修改。