短信验证码登录业务逻辑
校验账号是否存在 根据前端传过来的值查询数据库,查不到的时候直接抛出异常 提示手机号错误 返回给前端
Redis 校验短袖验证码 拼接手机号对应的Redis 验证码缓存key，读取缓存里存的验证码 ，读取缓存key 当没读取缓存时，提示验证码过期报错 ， 缓存读取到的验证码需要和用户输入的验证码保持一致 ，不一致提示验证码错误 。
登录通过时生成token令牌 用求职责ID、手机号生成access_token短期访问令牌 refresh_token 刷新令牌打印日志记录两个token
清扫缓存并返回信息 删除 Redis 你已使用过验证码 防止重复使用
@staticmethod async def login_by_mobile_code(login_request: LoginSmsCodeRequest): """ 1:判断手机号是否存在 2:redis中是否存在验证码并且一样,登录成功,生成token,删除redis中的验证码 3:不一样,登录失败 :return: """ jobseeker = await JobSeeker.get_or_none(mobile=login_request.mobile) if jobseeker is None: raise Exception("手机号不存在") key = f"boss-api:login:sms:{login_request.mobile}" redis_code = redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != login_request.code: raise Exception("验证码错误") 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}") redis_client.delete(key) return { "jobseeker": jobseeker, "access_token": access_token, "refresh_token": refresh_token, }将求职责的头像和前端传来的图片转换成二进制，调用阿里云存入到求职责的文件夹里面，上传成功后判断执行结果 ，上传成功打印日志 把oss返回的图片生成一个链接赋值给用户头像，最终返回上传资源信息 ，文件上传失败则抛出异常传过前端头像上传失败
async def update_avatar(job_seeker_id: str, file:UploadFile): file_content = await file.read() filename = file.filename oss = AliyunOSSTool() is_success, success_res = oss.upload_single_file( file_content, filename, oss_path="job_seeker_avatar/"#里面填写阿里云的文件名称 ) if is_success: logger.info("上传成功") jobseeker = await JobSeeker.get_or_none(id = job_seeker_id) jobseeker.avatar = success_res["access_url"] await jobseeker.save() return success_res else: raise Exception("上传失败")该静态异步方法实现求职者短信验证码注册功能，接收手机号、验证码、密码注册参数。首先读取 Redis 中该手机号缓存的验证码，无缓存则提示验证码过期，缓存与输入不一致则返回验证码错误；接着查询数据库判断手机号是否已注册，存在账号则提示直接登录。校验全部通过后自动创建求职者账号，默认昵称由手机号后四位生成，同时清除 Redis 内已使用的验证码。最后根据新用户 ID 与手机号生成访问令牌和刷新令牌，直接返回注册成功提示与双 token，用户完成注册后无需再次登录，简化注册登录流程。
@staticmethod async def register(register_req: RegisterRequest): mobile = register_req.mobile code = register_req.code password = register_req.password # 1.校验验证码 key = f"boss-api:login:sms:{mobile}" redis_code = await redis_client.get(key) if not redis_code: raise Exception("验证码已过期，请重新获取") if redis_code != code: raise Exception("验证码错误") # 2.校验手机号是否已注册 exist_user = await JobSeeker.get_or_none(mobile=mobile) if exist_user: raise Exception("该手机号已经注册，直接登录") # 3.创建用户 new_user = await JobSeeker.create( mobile=mobile, password=password, nickname=f"用户{mobile[-4:]}" ) # 删除验证码 await redis_client.delete(key) # 4.登录直接返回token access_token, refresh_token = create_tokens(str(new_user.id), new_user.mobile) return { "msg": "注册成功", "access_token": access_token, "refresh_token": refresh_token }密码重置功能实现分析
该代码片段实现了一个基于手机验证码的密码重置功能，属于静态异步方法。以下是核心逻辑分解：
参数接收与验证
方法接收ResetPwdRequest类型的参数，包含三个关键字段：
mobile = reset_req.mobile # 手机号 code = reset_req.code # 短信验证码 new_pwd = reset_req.new_password # 新密码验证码校验流程
从Redis获取存储的验证码进行比对：
key = f"boss-api:login:sms:{mobile}" redis_code = await redis_client.get(key) if not redis_code: raise Exception("验证码已过期，请重新获取") if redis_code != code: raise Exception("验证码错误")使用get_or_none方法避免未注册用户修改密码。
密码更新操作
更新用户对象密码字段并持久化
user.password = new_pwd await user.save() await redis_client.delete(key) # 清除已使用的验证码采用明文存储密码，实际生产环境应使用哈希加密（如bcrypt）。
安全增强建议
密码加密存储：
# 使用密码哈希示例 from passlib.hash import bcrypt user.password = bcrypt.hash(new_pwd)验证码时效控制： 建议在Redis设置时添加过期时间（如300秒），避免手动清理：
await redis_client.setex(key, 300, code)异常处理优化 使用自定义异常类代替通用Exception，提高代码可读性和可维护性。
class MobileNotExistError(Exception): pass class PasswordError(Exception): pass密码安全增强 建议使用哈希加密存储和验证密码，而不是明文比较。
from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # 密码验证改为 if not pwd_context.verify(login_mobile_request.password, jobseeker.password): raise PasswordError("密码错误")日志记录改进 建议将敏感信息如token进行脱敏处理后再记录。
logger.info(f"生成 access_token:{access_token[:5]}...") logger.info(f"生成 refresh_token:{refresh_token[:5]}...")返回数据结构可以考虑使用Pydantic模型规范返回数据结构。
class LoginResponse(BaseModel): jobseeker: JobSeeker access_token: str refresh_token: str完整优化版本
@staticmethod async def login_by_mobile_password(login_mobile_request: LoginMobileRequest): jobseeker = await JobSeeker.get_or_none(mobile=login_mobile_request.mobile) if jobseeker is None: raise MobileNotExistError("手机号不存在") if not pwd_context.verify(login_mobile_request.password, jobseeker.password): raise PasswordError("密码错误") access_token, refresh_token = create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f"生成 access_token:{access_token[:5]}...") logger.info(f"生成 refresh_token:{refresh_token[:5]}...") return LoginResponse( jobseeker=jobseeker, access_token=access_token, refresh_token=refresh_token )使用@fileupload_router.post装饰器定义了一个POST请求的路由，路径为/upload
添加了summary="上传文件"作为接口的简要描述
上传成功时记录日志并返回JSON响应：
{ "code": 1, "message": "上传成功", "data": success_res }