求职者端--手机号密码登录在之前项目中都做的是最基本的账密登录只进行了账号密码的比对这显然是不符合实际的。在本次加入了token更符合现实情况。首先将生成token和解析token的方法进行封装本方案采用 Access/Refresh 长短双 Token 架构登录时生成两类载荷携带type标识的 JWT 令牌access 短期有效用于业务鉴权refresh 长期有效仅负责刷新凭证封装通用校验方法强制区分令牌类型避免 Token 越权调用access 过期后依靠合法 refresh 令牌生成新访问令牌。该机制缩短有效凭证暴露窗口提升安全性但缺少 Refresh 黑名单与令牌轮换逻辑登出、改密场景下旧刷新令牌仍可正常使用生产环境建议搭配 Redis 黑名单优化。# 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_tokenstaticmethod async def login_by_mobile_password(login_mobile_request: LoginMobileRequest): jobseekerawait JobSeeker.get_or_none(mobilelogin_mobile_request.mobile) if jobseeker is None: raise Exception(手机号不存在) if jobseeker.password ! login_mobile_request.password: raise Exception(密码错误) access_token,refresh_token create_tokens(str(jobseeker.id), jobseeker.mobile) return { jobseeker: jobseeker, access_token: access_token, refresh_token: refresh_token }job_seeker_router.post(/login_by_mobile_password, summary手机号密码登录, description手机号密码登录) async def login_by_mobile_password(login_mobile_request:LoginMobileRequest): res await JobSeekerService.login_by_mobile_password(login_mobile_request) return { code:1, message:登录成功, data:res }业务 Service 层调用封装的create_tokens()使用求职者唯一标识签发 JWT 令牌并响应至前端后续接口依托令牌完成身份认证避免反复传递账号凭证实现无状态鉴权有效提升系统安全等级。求职者端--手机号验证码登录在此处封装了生成验证码的方法方便之后直接调用在这里使用的是简单的6位随机数其实是不太符合实际需求的如果专业一点可以直接调用容联云的短信发送。def get_random_code(length: int 4): code .join(random.choices(1324567890, klength)) return codestaticmethod async def send_sms_code(mobile: str): jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) codeget_random_code(length6) logger.info(f生成验证码:{code}) key fboss-api:login:sms:{mobile} redis_client.set(key, code, ex60 * 2) return codejob_seeker_router.post(/send_sms_code/{mobile}, summary发送验证码, description发送验证码) async def send_sms_code(mobile: str): await JobSeekerService.send_sms_code(mobile) return { code: 1, message: 发送成功, }因为验证码是有时效性的只有在规定时间内才有效在此处显然缓存十分有效。staticmethod async def login_by_mobile_code(login_request: LoginSmsCodeRequest): 1:判断手机号是否存在 2:redis中是否存在验证码并且一样,登录成功,生成token,删除redis中的验证码 3:不一样,登录失败 :return: jobseeker await JobSeeker.get_or_none(mobilelogin_request.mobile) if jobseeker is None: raise Exception(手机号不存在) key fboss-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, }job_seeker_router.post(/login_by_mobile_code, summary手机号验证码登录, description手机号验证码登录) async def login_by_mobile_code(login_request: LoginSmsCodeRequest): res await JobSeekerService.login_by_mobile_code(login_request) return { code: 1, message: 登录成功, data: res }在这里与手机号密码登录同理在对手机号和验证码的基础判断之后也是调用相同的方法生成token在生成token之后记得将缓存中对应的验证码删除即可。求职者端--注册staticmethod async def register(register_req: RegisterRequest): mobile register_req.mobile code register_req.code password register_req.password key fboss-api:login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! code: raise Exception(验证码错误) jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is not None: raise Exception(手机号已存在,请登录) new_jobseeker await JobSeeker.create( mobilemobile, passwordpassword ) await redis_client.delete(key) access_token, refresh_token create_tokens(str(new_jobseeker.id), new_jobseeker.mobile) return { jobseeker: new_jobseeker, access_token: access_token, refresh_token: refresh_token, }job_seeker_router.post(/register, summary注册, description注册) async def register(register_req: RegisterRequest): res await JobSeekerService.register(register_req) return { code: 1, message: 注册成功, data: res }注册类似于手机号密码登录与手机号验证码登录的结合首先获取前端传过来的手机号验证码密码等信息在验证码验证的时候去缓存里与验证码进行对比在进行基础型的判断之后删除对应缓存创建新用户并且调用与登录一样的方法生成token。求职者端--忘记密码/重置密码job_seeker_router.post(/reset_pwd, summary忘记密码重置接口) async def reset_password(reset_req: ResetPwdRequest): await JobSeekerService.reset_password(reset_req) return { code: 1, message: 重置成功 }staticmethod async def reset_password(reset_req: ResetPwdRequest): mobile reset_req.mobile code reset_req.code new_pwdreset_req.new_password key fboss-api:login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! code: raise Exception(验证码错误) jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) jobseeker.password new_pwd await jobseeker.save() await redis_client.delete(key) return 1忘记密码的本质就是orm的修改接受前端传过来的手机号验证码新密码的信息对验证码进行判断后将新密码替换掉旧密码即可删除对应缓存。求职者端-我的页面获取个人信息# 获取用户信息 staticmethod async def get_job_seeker_info_by_id(job_seeker_id: str): job_seekerawait JobSeeker.get_or_none(idjob_seeker_id) job_seeker.nicknamef用户{job_seeker.mobile} if job_seeker.nickname is None else job_seeker.nickname job_seeker.avatar https://gips1.baidu.com/it/u436886321,1020119268fm3028app3028fJPEGfmtauto?w1280h960 if job_seeker.avatar is None else job_seeker.avatar return job_seekerjob_seeker_router.get(/me, summary获取当前用户信息, description获取当前用户信息) async def get_me(job_seeker_idDepends(get_job_seeker_info)): job_seeker await JobSeekerService.get_job_seeker_info_by_id(job_seeker_id) return { code: 1, message: 获取成功, data: job_seeker }接口借助依赖注入解析请求头中的 Token 提取求职者 ID调用 Service 层依据 ID 查询并加工用户数据完成登录求职者个人信息的拉取。如下时封装的依赖注入方法def get_job_seeker_info(authorization: str|None Header(None)) - str: if not authorization: raise Exception(未登录) tokenauthorization.replace(Bearer ,) payloadverify_token(token) if payload is None: raise Exception(验证失败) job_seeker_idpayload[user_id] return job_seeker_id求职者端-上传/更新头像async def update_avatar(job_seeker_id: str,file:UploadFile): file_contentawait file.read() filenamefile.filename ossAliyunOSSTool() is_success,success_resoss.upload_single_file(file_content,filename,oss_pathjob_seeker_avatar/) if is_success: jobseekerawait JobSeeker.get_or_none(idjob_seeker_id) jobseeker.avatarsuccess_res[access_url] await jobseeker.save() return success_res else: raise Exception(上传失败)job_seeker_router.post(/update_avatar, summary上传头像, description上传头像) async def update_avatar(file: UploadFile File(...,description上传文件), job_seeker_idDepends(get_job_seeker_info)): await JobSeekerService.update_avatar(job_seeker_id,file) return { code: 1, message: 上传头像成功 }因为用户不论身份量是十分庞大的放置到本地磁盘显然是不现实的在此处接入了阿里云的对象存储具体配置可以访问阿里云官方根据SDK进行操作。上传文件的接口通过依赖注入解析请求头 Token校验身份并获取当前登录求职者 ID接收前端提交的头像文件在 Service 层调用封装好的阿里云 OSS 工具将图片二进制流上传至云端对象存储。文件上传成功后获取公开访问 URL查询对应求职者记录并更新 avatar 字段执行数据持久化若 OSS 上传流程出现异常主动抛出业务异常交由全局异常处理器统一响应错误信息。总结求职者模块实现注册、多方式登录、密码重置、个人信息查询与头像上传能力基于 JWT 长短双 Token 完成无状态身份认证利用 Redis 维护短信验证码时效接入阿里云 OSS 处理文件存储依靠依赖注入统一鉴权并遵循路由、Service 分层架构规范开发。
仿BOSS招聘平台实现（1）
求职者端--手机号密码登录在之前项目中都做的是最基本的账密登录只进行了账号密码的比对这显然是不符合实际的。在本次加入了token更符合现实情况。首先将生成token和解析token的方法进行封装本方案采用 Access/Refresh 长短双 Token 架构登录时生成两类载荷携带type标识的 JWT 令牌access 短期有效用于业务鉴权refresh 长期有效仅负责刷新凭证封装通用校验方法强制区分令牌类型避免 Token 越权调用access 过期后依靠合法 refresh 令牌生成新访问令牌。该机制缩短有效凭证暴露窗口提升安全性但缺少 Refresh 黑名单与令牌轮换逻辑登出、改密场景下旧刷新令牌仍可正常使用生产环境建议搭配 Redis 黑名单优化。# 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_tokenstaticmethod async def login_by_mobile_password(login_mobile_request: LoginMobileRequest): jobseekerawait JobSeeker.get_or_none(mobilelogin_mobile_request.mobile) if jobseeker is None: raise Exception(手机号不存在) if jobseeker.password ! login_mobile_request.password: raise Exception(密码错误) access_token,refresh_token create_tokens(str(jobseeker.id), jobseeker.mobile) return { jobseeker: jobseeker, access_token: access_token, refresh_token: refresh_token }job_seeker_router.post(/login_by_mobile_password, summary手机号密码登录, description手机号密码登录) async def login_by_mobile_password(login_mobile_request:LoginMobileRequest): res await JobSeekerService.login_by_mobile_password(login_mobile_request) return { code:1, message:登录成功, data:res }业务 Service 层调用封装的create_tokens()使用求职者唯一标识签发 JWT 令牌并响应至前端后续接口依托令牌完成身份认证避免反复传递账号凭证实现无状态鉴权有效提升系统安全等级。求职者端--手机号验证码登录在此处封装了生成验证码的方法方便之后直接调用在这里使用的是简单的6位随机数其实是不太符合实际需求的如果专业一点可以直接调用容联云的短信发送。def get_random_code(length: int 4): code .join(random.choices(1324567890, klength)) return codestaticmethod async def send_sms_code(mobile: str): jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) codeget_random_code(length6) logger.info(f生成验证码:{code}) key fboss-api:login:sms:{mobile} redis_client.set(key, code, ex60 * 2) return codejob_seeker_router.post(/send_sms_code/{mobile}, summary发送验证码, description发送验证码) async def send_sms_code(mobile: str): await JobSeekerService.send_sms_code(mobile) return { code: 1, message: 发送成功, }因为验证码是有时效性的只有在规定时间内才有效在此处显然缓存十分有效。staticmethod async def login_by_mobile_code(login_request: LoginSmsCodeRequest): 1:判断手机号是否存在 2:redis中是否存在验证码并且一样,登录成功,生成token,删除redis中的验证码 3:不一样,登录失败 :return: jobseeker await JobSeeker.get_or_none(mobilelogin_request.mobile) if jobseeker is None: raise Exception(手机号不存在) key fboss-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, }job_seeker_router.post(/login_by_mobile_code, summary手机号验证码登录, description手机号验证码登录) async def login_by_mobile_code(login_request: LoginSmsCodeRequest): res await JobSeekerService.login_by_mobile_code(login_request) return { code: 1, message: 登录成功, data: res }在这里与手机号密码登录同理在对手机号和验证码的基础判断之后也是调用相同的方法生成token在生成token之后记得将缓存中对应的验证码删除即可。求职者端--注册staticmethod async def register(register_req: RegisterRequest): mobile register_req.mobile code register_req.code password register_req.password key fboss-api:login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! code: raise Exception(验证码错误) jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is not None: raise Exception(手机号已存在,请登录) new_jobseeker await JobSeeker.create( mobilemobile, passwordpassword ) await redis_client.delete(key) access_token, refresh_token create_tokens(str(new_jobseeker.id), new_jobseeker.mobile) return { jobseeker: new_jobseeker, access_token: access_token, refresh_token: refresh_token, }job_seeker_router.post(/register, summary注册, description注册) async def register(register_req: RegisterRequest): res await JobSeekerService.register(register_req) return { code: 1, message: 注册成功, data: res }注册类似于手机号密码登录与手机号验证码登录的结合首先获取前端传过来的手机号验证码密码等信息在验证码验证的时候去缓存里与验证码进行对比在进行基础型的判断之后删除对应缓存创建新用户并且调用与登录一样的方法生成token。求职者端--忘记密码/重置密码job_seeker_router.post(/reset_pwd, summary忘记密码重置接口) async def reset_password(reset_req: ResetPwdRequest): await JobSeekerService.reset_password(reset_req) return { code: 1, message: 重置成功 }staticmethod async def reset_password(reset_req: ResetPwdRequest): mobile reset_req.mobile code reset_req.code new_pwdreset_req.new_password key fboss-api:login:sms:{mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! code: raise Exception(验证码错误) jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) jobseeker.password new_pwd await jobseeker.save() await redis_client.delete(key) return 1忘记密码的本质就是orm的修改接受前端传过来的手机号验证码新密码的信息对验证码进行判断后将新密码替换掉旧密码即可删除对应缓存。求职者端-我的页面获取个人信息# 获取用户信息 staticmethod async def get_job_seeker_info_by_id(job_seeker_id: str): job_seekerawait JobSeeker.get_or_none(idjob_seeker_id) job_seeker.nicknamef用户{job_seeker.mobile} if job_seeker.nickname is None else job_seeker.nickname job_seeker.avatar https://gips1.baidu.com/it/u436886321,1020119268fm3028app3028fJPEGfmtauto?w1280h960 if job_seeker.avatar is None else job_seeker.avatar return job_seekerjob_seeker_router.get(/me, summary获取当前用户信息, description获取当前用户信息) async def get_me(job_seeker_idDepends(get_job_seeker_info)): job_seeker await JobSeekerService.get_job_seeker_info_by_id(job_seeker_id) return { code: 1, message: 获取成功, data: job_seeker }接口借助依赖注入解析请求头中的 Token 提取求职者 ID调用 Service 层依据 ID 查询并加工用户数据完成登录求职者个人信息的拉取。如下时封装的依赖注入方法def get_job_seeker_info(authorization: str|None Header(None)) - str: if not authorization: raise Exception(未登录) tokenauthorization.replace(Bearer ,) payloadverify_token(token) if payload is None: raise Exception(验证失败) job_seeker_idpayload[user_id] return job_seeker_id求职者端-上传/更新头像async def update_avatar(job_seeker_id: str,file:UploadFile): file_contentawait file.read() filenamefile.filename ossAliyunOSSTool() is_success,success_resoss.upload_single_file(file_content,filename,oss_pathjob_seeker_avatar/) if is_success: jobseekerawait JobSeeker.get_or_none(idjob_seeker_id) jobseeker.avatarsuccess_res[access_url] await jobseeker.save() return success_res else: raise Exception(上传失败)job_seeker_router.post(/update_avatar, summary上传头像, description上传头像) async def update_avatar(file: UploadFile File(...,description上传文件), job_seeker_idDepends(get_job_seeker_info)): await JobSeekerService.update_avatar(job_seeker_id,file) return { code: 1, message: 上传头像成功 }因为用户不论身份量是十分庞大的放置到本地磁盘显然是不现实的在此处接入了阿里云的对象存储具体配置可以访问阿里云官方根据SDK进行操作。上传文件的接口通过依赖注入解析请求头 Token校验身份并获取当前登录求职者 ID接收前端提交的头像文件在 Service 层调用封装好的阿里云 OSS 工具将图片二进制流上传至云端对象存储。文件上传成功后获取公开访问 URL查询对应求职者记录并更新 avatar 字段执行数据持久化若 OSS 上传流程出现异常主动抛出业务异常交由全局异常处理器统一响应错误信息。总结求职者模块实现注册、多方式登录、密码重置、个人信息查询与头像上传能力基于 JWT 长短双 Token 完成无状态身份认证利用 Redis 维护短信验证码时效接入阿里云 OSS 处理文件存储依靠依赖注入统一鉴权并遵循路由、Service 分层架构规范开发。
