一、分析实现功能

需要实现以下功能：

账号密码登录

手机号+验证码登录

发送验证码

第三方登录-钉钉登录,

绑定手机号, 识别同一个用户

注册

查询个人信息

二、数据库设计

t_job_seeker求职者表

from tortoise.models import Model from tortoise import fields class JobSeeker(Model): """求职者表 t_job_seeker""" # 主键ID，自增 id = fields.IntField(pk=True, description="主键ID,自增") # 手机号，唯一 mobile = fields.CharField(max_length=20, unique=True, description="手机号") # 密码 password = fields.CharField(max_length=255,null=True, description="密码") # 头像（存储图片url） avatar = fields.CharField(max_length=500, null=True, description="头像地址") # 昵称 nickname = fields.CharField(max_length=100, null=True, description="昵称") # 邮箱 email = fields.CharField(max_length=100, null=True, description="邮箱") # 微信号 wechat = fields.CharField(max_length=100, null=True, description="微信号") # 创建时间 create_time = fields.DatetimeField(auto_now_add=True, description="创建时间") # 修改时间 update_time = fields.DatetimeField(auto_now=True, description="修改时间") class Meta: table = "t_job_seeker" # 指定数据库表名 table_description = "求职者表"

三、接口设计与开发

下载依赖

pip install python-jose

创建工具类:app/utils/jwt_util.py。用于使用FastAPI生成Token

from jose import jwt, JWTError from datetime import datetime, timedelta from typing import Optional, Tuple from app.config.settings import settings SECRET_KEY = settings.secret_key ALGORITHM = settings.ALGORITHM # 配置不同令牌的过期时间（建议在settings中配置） ACCESS_TOKEN_EXPIRE_MINUTES = settings.token_expire_minutes # 如30分钟 REFRESH_TOKEN_EXPIRE_DAYS = settings.REFRESH_TOKEN_EXPIRE_DAYS # 如7天 # 1. 生成access_token和refresh_token def create_tokens(user_id: str, username: str) -> Tuple[str, str]: """ 生成访问令牌和刷新令牌 :param user_id: 用户ID :param username: 用户名 :return: (access_token, refresh_token) """ # 构建公共载荷（可根据需求扩展） common_payload = { "user_id": user_id, "username": username, "type": "access" # 标记令牌类型 } # 1. 生成access_token（短期） access_payload = common_payload.copy() access_payload["exp"] = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM) # 2. 生成refresh_token（长期） refresh_payload = common_payload.copy() refresh_payload["type"] = "refresh" # 标记为刷新令牌 refresh_payload["exp"] = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS) refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm=ALGORITHM) return access_token, refresh_token # 2. 通用令牌验证（区分令牌类型） def verify_token(token: str, token_type: str = "access") -> Optional[dict]: """ 验证令牌并解析载荷 :param token: 令牌字符串 :param token_type: 期望的令牌类型（access/refresh） :return: 解析后的载荷（验证失败返回None） """ try: payload = jwt.decode( token, SECRET_KEY, algorithms=[ALGORITHM], options={"verify_exp": True} # 强制校验过期时间 ) # 校验令牌类型 if payload.get("type") != token_type: print(f"验证失败：令牌类型不是{token_type}") return None # 校验必要字段 if not all(key in payload for key in ["user_id", "username"]): print("验证失败：载荷缺少必要字段") return None return payload except JWTError as e: print(f"验证失败：{e}") return None # 3. 使用refresh_token刷新access_token def refresh_access_token(refresh_token: str) -> Optional[str]: """ 通过有效的refresh_token生成新的access_token :param refresh_token: 刷新令牌 :return: 新的access_token（刷新失败返回None） """ # 先验证refresh_token的有效性 payload = verify_token(refresh_token, token_type="refresh") if not payload: return None # 从refresh_token的载荷中提取用户信息，生成新的access_token user_id = payload["user_id"] username = payload["username"] # 构建新的access_token载荷 access_payload = { "user_id": user_id, "username": username, "type": "access", "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) } new_access_token = jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM) return new_access_token

在service 调用一下:

# 生成token access_token, refresh_token = create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f"生成 access_token:{access_token}") logger.info(f"生成 refresh_token:{refresh_token}")

创建工具类:app/utils/generate_code_util.生成手机号验证码

""" 生成随机数的验证码,可以自定义位数 """ import random def get_random_code(length: int = 4): code = ''.join(random.choices('1324567890', k=length)) return code

创建app/apis/job_seeker_api.py。书写接口

from fastapi import APIRouter, Header, Depends, UploadFile, File from app.core.depends import get_server_info from app.schemas.job_seeker import ( LoginMobileRequest, LoginSmsCodeRequest, RegisterRequest, ForgotPasswordRequest, ) from app.services.job_seeker import LoginMobileService, JobSeekerService from app.core.logging import logger job_seeker_router = APIRouter(prefix="/job_seeker", tags=["用户管理"]) @job_seeker_router.post("/login", summary="手机号登录") async def login_mobile(mobiles: LoginMobileRequest): res = await LoginMobileService.login_mobile(mobiles) return {"code": 1, "msg": "登陆成功", "data": res}

创建app/services/job_seek.py。用于书写逻辑

import redis from fastapi import UploadFile from app.core.depends import get_server_info from app.models import JobSeeker from app.schemas.job_seeker import ( LoginMobileRequest, LoginSmsCodeRequest, RegisterRequest, ForgotPasswordRequest, ) from app.utils.generate_code_util import get_random_code from app.utils.jwt_util import create_tokens from app.core.logging import logger from app.utils.oss_util import AliyunOSSTool redis_client = redis.Redis(host="127.0.0.1", port=6379, db=9, decode_responses=True) class LoginMobileService: @staticmethod async def login_mobile(mobiles: LoginMobileRequest): jobseeker = await JobSeeker.get_or_none(mobile=mobiles.mobile) if jobseeker is None: raise Exception("用户不存在") if jobseeker.password != mobiles.password: raise Exception("密码错误") # 生成token access_token, refresh_token = create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f"access_token: {access_token}") logger.info(f"refresh_token: {refresh_token}") return {"jobseeker": jobseeker, "access_token": access_token, "refresh_token": refresh_token}

创建app/schemas.py.。 创建请求体。

from pydantic import BaseModel, Field class LoginMobileRequest(BaseModel): mobile: str = Field(..., description="手机号") password: str = Field(..., description="密码")

四、总结

这套 FastAPI + TortoiseORM 技术栈上手简单。实现简单的手机号登录。