1. 功能概述在现代Web应用中登录注册是基础且核心的功能模块。本文将基于Python FastAPI框架实现一套完整的用户认证系统包含以下功能手机号密码登录传统账号密码登录方式手机号验证码登录短信验证码登录更安全便捷用户注册手机号验证码注册自动登录忘记密码通过验证码重置密码JWT令牌管理使用双令牌机制access_token refresh_token2. 技术栈与依赖2.1 核心依赖# requirements.txtfastapi0.104.1uvicorn0.24.0python-jose[cryptography]3.3.0redis5.0.1tortoise-orm0.20.0pydantic2.5.0python-multipart0.0.62.2 安装命令pipinstallfastapi uvicorn python-jose[cryptography]redis tortoise-orm pydantic python-multipart3. 数据库模型设计3.1 用户模型# app/models/jobseeker.pyfromtortoiseimportfields,modelsfromtortoise.contrib.pydanticimportpydantic_model_creatorclassJobSeeker(models.Model):求职者用户模型idfields.IntField(pkTrue)mobilefields.CharField(max_length11,uniqueTrue,description手机号)passwordfields.CharField(max_length255,description密码加密存储)nicknamefields.CharField(max_length50,nullTrue,description昵称)avatarfields.CharField(max_length255,nullTrue,description头像URL)created_atfields.DatetimeField(auto_now_addTrue)updated_atfields.DatetimeField(auto_nowTrue)classMeta:tablejob_seekerdef__str__(self):returnf{self.nickname}({self.mobile})# Pydantic模型JobSeeker_Pydanticpydantic_model_creator(JobSeeker,nameJobSeeker)JobSeekerIn_Pydanticpydantic_model_creator(JobSeeker,nameJobSeekerIn,exclude_readonlyTrue)4. 请求与响应模型4.1 登录相关模型# app/schemas/auth.pyfrompydanticimportBaseModel,Field,validatorimportreclassLoginMobileRequest(BaseModel):手机号密码登录请求mobile:strField(...,min_length11,max_length11,description手机号)password:strField(...,min_length6,max_length20,description密码)validator(mobile)defvalidate_mobile(cls,v):ifnotre.match(r^1[3-9]\d{9}$,v):raiseValueError(手机号格式不正确)returnvclassLoginSMSRequest(BaseModel):手机号验证码登录请求mobile:strField(...,min_length11,max_length11,description手机号)code:strField(...,min_length4,max_length6,description验证码)validator(mobile)defvalidate_mobile(cls,v):ifnotre.match(r^1[3-9]\d{9}$,v):raiseValueError(手机号格式不正确)returnvclassRegisterRequest(BaseModel):用户注册请求mobile:strField(...,min_length11,max_length11,description手机号)password:strField(...,min_length6,max_length20,description密码)code:strField(...,min_length4,max_length6,description验证码)validator(mobile)defvalidate_mobile(cls,v):ifnotre.match(r^1[3-9]\d{9}$,v):raiseValueError(手机号格式不正确)returnvclassResetPwdRequest(BaseModel):重置密码请求mobile:strField(...,min_length11,max_length11,description手机号)new_password:strField(...,min_length6,max_length20,description新密码)code:strField(...,min_length4,max_length6,description验证码)validator(mobile)defvalidate_mobile(cls,v):ifnotre.match(r^1[3-9]\d{9}$,v):raiseValueError(手机号格式不正确)returnvclassTokenResponse(BaseModel):令牌响应access_token:strField(...,description访问令牌)refresh_token:strField(...,description刷新令牌)token_type:strField(defaultbearer,description令牌类型)expires_in:intField(...,description过期时间秒)5. JWT令牌工具类5.1 安装JWT依赖pipinstallpython-jose[cryptography]5.2 JWT工具类实现# app/utils/jwt_util.pyfromjoseimportjwt,JWTErrorfromdatetimeimportdatetime,timedeltafromtypingimportOptional,Tuplefromapp.config.settingsimportsettingsimportlogging loggerlogging.getLogger(__name__)SECRET_KEYsettings.secret_key ALGORITHMsettings.ALGORITHM# 配置不同令牌的过期时间ACCESS_TOKEN_EXPIRE_MINUTESsettings.token_expire_minutes# 如30分钟REFRESH_TOKEN_EXPIRE_DAYSsettings.REFRESH_TOKEN_EXPIRE_DAYS# 如7天defcreate_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_payloadcommon_payload.copy()access_payload[exp]datetime.utcnow()timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES)access_tokenjwt.encode(access_payload,SECRET_KEY,algorithmALGORITHM)# 2. 生成refresh_token长期refresh_payloadcommon_payload.copy()refresh_payload[type]refresh# 标记为刷新令牌refresh_payload[exp]datetime.utcnow()timedelta(daysREFRESH_TOKEN_EXPIRE_DAYS)refresh_tokenjwt.encode(refresh_payload,SECRET_KEY,algorithmALGORITHM)returnaccess_token,refresh_tokendefverify_token(token:str,token_type:straccess)-Optional[dict]: 验证令牌并解析载荷 :param token: 令牌字符串 :param token_type: 期望的令牌类型access/refresh :return: 解析后的载荷验证失败返回None try:payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM],options{verify_exp:True}# 强制校验过期时间)# 校验令牌类型ifpayload.get(type)!token_type:logger.warning(f验证失败令牌类型不是{token_type})returnNone# 校验必要字段ifnotall(keyinpayloadforkeyin[user_id,username]):logger.warning(验证失败载荷缺少必要字段)returnNonereturnpayloadexceptJWTErrorase:logger.error(fJWT验证失败{e})returnNonedefrefresh_access_token(refresh_token:str)-Optional[str]: 通过有效的refresh_token生成新的access_token :param refresh_token: 刷新令牌 :return: 新的access_token刷新失败返回None # 先验证refresh_token的有效性payloadverify_token(refresh_token,token_typerefresh)ifnotpayload:returnNone# 从refresh_token的载荷中提取用户信息生成新的access_tokenuser_idpayload[user_id]usernamepayload[username]# 构建新的access_token载荷access_payload{user_id:user_id,username:username,type:access,exp:datetime.utcnow()timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES)}new_access_tokenjwt.encode(access_payload,SECRET_KEY,algorithmALGORITHM)returnnew_access_token5.3 配置文件# app/config/settings.pyfrompydantic_settingsimportBaseSettingsclassSettings(BaseSettings):# 应用配置app_name:strBoss直聘APIdebug:boolTrue# JWT配置secret_key:stryour-secret-key-change-in-production# 生产环境请使用强密钥ALGORITHM:strHS256token_expire_minutes:int30# access_token过期时间分钟REFRESH_TOKEN_EXPIRE_DAYS:int7# refresh_token过期时间天# Redis配置redis_host:strlocalhostredis_port:int6379redis_db:int0redis_password:str# 数据库配置database_url:strsqlite://db.sqlite3classConfig:env_file.envsettingsSettings()6. 验证码工具类6.1 随机验证码生成# app/utils/code_util.pyimportrandomimportstringdefget_random_code(length:int4)-str: 生成随机数字验证码 :param length: 验证码长度默认4位 :return: 随机验证码字符串 # 使用数字生成验证码code.join(random.choices(0123456789,klength))returncodedefget_random_alphanumeric(length:int6)-str: 生成随机字母数字验证码 :param length: 验证码长度默认6位 :return: 随机验证码字符串 # 使用字母和数字生成验证码charactersstring.ascii_lettersstring.digits code.join(random.choices(characters,klength))returncode7. Redis缓存配置# app/core/redis.pyimportredis.asyncioasredisfromapp.config.settingsimportsettingsimportlogging loggerlogging.getLogger(__name__)# Redis连接池redis_clientNoneasyncdefinit_redis():初始化Redis连接globalredis_clienttry:redis_clientredis.Redis(hostsettings.redis_host,portsettings.redis_port,dbsettings.redis_db,passwordsettings.redis_passwordifsettings.redis_passwordelseNone,decode_responsesTrue,# 自动解码为字符串socket_connect_timeout5,socket_keepaliveTrue)# 测试连接awaitredis_client.ping()logger.info(Redis连接成功)exceptExceptionase:logger.error(fRedis连接失败:{e})raiseasyncdefclose_redis():关闭Redis连接globalredis_clientifredis_client:awaitredis_client.close()logger.info(Redis连接已关闭)8. 业务服务层实现8.1 用户服务类# app/services/jobseeker_service.pyfromapp.models.jobseekerimportJobSeekerfromapp.schemas.authimport(LoginMobileRequest,LoginSMSRequest,RegisterRequest,ResetPwdRequest,TokenResponse)fromapp.utils.jwt_utilimportcreate_tokens,verify_tokenfromapp.utils.code_utilimportget_random_codefromapp.core.redisimportredis_clientfromfastapiimportHTTPExceptionimportlogging loggerlogging.getLogger(__name__)classJobSeekerService:求职者服务类staticmethodasyncdeflogin_by_mobile_password(login_mobile_request:LoginMobileRequest): 手机号密码登录 思路 1. 查询手机号是否存在 2. 验证密码是否正确 3. 生成JWT令牌 # 1. 查询用户是否存在jobseekerawaitJobSeeker.get_or_none(mobilelogin_mobile_request.mobile)ifjobseekerisNone:raiseHTTPException(status_code400,detail手机号不存在)# 2. 验证密码实际项目中密码应该加密存储ifjobseeker.password!login_mobile_request.password:raiseHTTPException(status_code400,detail密码错误)# 3. 生成JWT令牌access_token,refresh_tokencreate_tokens(str(jobseeker.id),jobseeker.mobile)logger.info(f生成 access_token:{access_token})logger.info(f生成 refresh_token:{refresh_token})return{code:1,message:登录成功,data:{jobseeker:jobseeker,access_token:access_token,refresh_token:refresh_token}}staticmethodasyncdefsend_login_sms_code(mobile:str): 发送登录短信验证码 思路 1. 校验手机号是否已注册登录必须已注册 2. 生成随机验证码 3. 存储到Redis设置2分钟过期 4. 实际项目中这里应该调用短信服务发送验证码 # 1. 登录获取验证码前必须校验手机号存在userawaitJobSeeker.get_or_none(mobilemobile)ifnotuser:raiseHTTPException(status_code400,detail该手机号尚未注册请先注册)# 2. 生成6位随机验证码codeget_random_code(length6)logger.info(f【登录验证码】mobile:{mobile}, code:{code})# 3. 存储到Redis设置2分钟过期keyfboss-api:login:sms:{mobile}awaitredis_client.set(key,code,ex60*2)# 4. 实际项目中这里应该调用短信服务发送验证码# await send_sms(mobile, f您的登录验证码是{code}2分钟内有效)returncodestaticmethodasyncdeflogin_by_sms(login_sms_request:LoginSMSRequest): 手机号验证码登录 思路 1. 从Redis获取验证码 2. 验证验证码是否正确 3. 查询用户信息 4. 生成JWT令牌 5. 删除Redis中的验证码 mobilelogin_sms_request.mobile codelogin_sms_request.code# 1. 从Redis获取验证码keyfboss-api:login:sms:{mobile}redis_codeawaitredis_client.get(key)ifnotredis_code:raiseHTTPException(status_code400,detail验证码已过期请重新获取)ifredis_code!code:raiseHTTPException(status_code400,detail验证码错误)# 2. 查询用户信息userawaitJobSeeker.get_or_none(mobilemobile)ifnotuser:raiseHTTPException(status_code400,detail用户不存在)# 3. 生成JWT令牌access_token,refresh_tokencreate_tokens(str(user.id),user.mobile)# 4. 删除Redis中的验证码防止重复使用awaitredis_client.delete(key)return{code:1,message:登录成功,data:{jobseeker:user,access_token:access_token,refresh_token:refresh_token}}staticmethodasyncdefsend_register_sms_code(mobile:str): 发送注册短信验证码 思路 1. 生成随机验证码 2. 存储到Redis设置2分钟过期 3. 注册不需要先校验手机号是否存在 # 1. 生成6位随机验证码codeget_random_code(length6)logger.info(f【注册验证码】mobile:{mobile}, code:{code})# 2. 存储到Redis设置2分钟过期keyfboss-api:register:sms:{mobile}awaitredis_client.set(key,code,ex60*2)# 3. 实际项目中这里应该调用短信服务发送验证码# await send_sms(mobile, f您的注册验证码是{code}2分钟内有效)returncodestaticmethodasyncdefregister(register_req:RegisterRequest): 用户注册 思路