一、分析实现功能需要实现以下功能账号密码登录手机号验证码登录发送验证码第三方登录-钉钉登录,绑定手机号, 识别同一个用户注册查询个人信息二、数据库设计t_job_seeker求职者表#字段名类型备注或者约束IDInt主键ID,自增手机号字符串唯一密码字符串头像字符串昵称字符串邮箱字符串微信号字符串创建时间日期时间修改时间日期时间from tortoise.models import Model from tortoise import fields class JobSeeker(Model): 求职者表 t_job_seeker # 主键ID自增 id fields.IntField(pkTrue, description主键ID,自增) # 手机号唯一 mobile fields.CharField(max_length20, uniqueTrue, description手机号) # 密码 password fields.CharField(max_length255,nullTrue, description密码) # 头像存储图片url avatar fields.CharField(max_length500, nullTrue, description头像地址) # 昵称 nickname fields.CharField(max_length100, nullTrue, description昵称) # 邮箱 email fields.CharField(max_length100, nullTrue, description邮箱) # 微信号 wechat fields.CharField(max_length100, nullTrue, description微信号) # 创建时间 create_time fields.DatetimeField(auto_now_addTrue, description创建时间) # 修改时间 update_time fields.DatetimeField(auto_nowTrue, description修改时间) class Meta: table t_job_seeker # 指定数据库表名 table_description 求职者表三、接口设计与开发下载依赖pip install python-jose创建工具类:app/utils/jwt_util.py。用于使用FastAPI生成Tokenfrom 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(minutesACCESS_TOKEN_EXPIRE_MINUTES) access_token jwt.encode(access_payload, SECRET_KEY, algorithmALGORITHM) # 2. 生成refresh_token长期 refresh_payload common_payload.copy() refresh_payload[type] refresh # 标记为刷新令牌 refresh_payload[exp] datetime.utcnow() timedelta(daysREFRESH_TOKEN_EXPIRE_DAYS) refresh_token jwt.encode(refresh_payload, SECRET_KEY, algorithmALGORITHM) 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_typerefresh) 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(minutesACCESS_TOKEN_EXPIRE_MINUTES) } new_access_token jwt.encode(access_payload, SECRET_KEY, algorithmALGORITHM) 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, klength)) 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(host127.0.0.1, port6379, db9, decode_responsesTrue) class LoginMobileService: staticmethod async def login_mobile(mobiles: LoginMobileRequest): jobseeker await JobSeeker.get_or_none(mobilemobiles.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(faccess_token: {access_token}) logger.info(frefresh_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 技术栈上手简单。实现简单的手机号登录。
