模拟BOSS招聘项目——求职者端功能实现-北京尧图网络

一、登录部分求职者表设计在models中添加job_seek.pyclass 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 求职者表为表添加模拟数据同时在项目的apisschemasservices中添加job_seek.py1、手机号密码登录apisjob_seeker_router APIRouter( prefix/job_seeker, tags[求职者] ) job_seeker_router.post(/login,summary用户登录) async def login_by_mobile_password(request:LoginByPhoneRequest): res await JobSeekerService.login_by_mobile_password(request) return { code: 1, msg: success, data: res }secvicesimport redis from app.models import JobSeeker 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( hostlocalhost, port6379, db6, decode_responsesTrue ) class JobSeekerService: staticmethod async def login_by_mobile_password(reqeuest): # 1. 验证手机号是否存在 jobseekerawait JobSeeker.get_or_none(mobilereqeuest.mobile) if jobseeker is None: raise Exception(手机号不存在) # 2. if jobseeker.password ! reqeuest.password: raise Exception(密码错误) # 3.登录成功返回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, }schemasclass LoginByPhoneRequest(BaseModel): mobile: str Field(..., min_length1, max_length20, description手机号) password: str Field(..., min_length6, max_length20, description密码)2、手机号验证码登录发送验证码apisjob_seeker_router.post(/send_sms_code/{mobile},summary发送验证码) async def send_sms_code(mobile:str): await JobSeekerService.send_sms_code(mobile) return { code:1,msg:success }servicesstaticmethod async def send_sms_code(mobile: str): jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) code get_random_code(length6) logger.info(f生成验证码:{code}) key fboss-api:login:sms:{mobile} redis_client.set(key, code, ex60 * 2) return codeutils.generate_code_util.py 生成随机数的验证码,可以自定义位数 import random def get_random_code(length: int 4): code .join(random.choices(1324567890, klength)) return code登录apisjob_seeker_router.post(/login_by_mobile_code,summary手机号验证码登录) async def login_by_mobile_code(request:LoginPhoneCodeRequest): res await JobSeekerService.login_by_mobile_code(request) return { code:1,msg:success,data:res }servicestaticmethod async def login_by_mobile_code(request): jobseeker await JobSeeker.get_or_none(mobilerequest.mobile) if jobseeker is None: raise Exception(手机号不存在) key fboss-api:login:sms:{request.mobile} redis_coderedis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! 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, }schemasclass LoginPhoneCodeRequest(BaseModel): mobile: str Field(..., min_length1, max_length20, description手机号) code: str Field(..., min_length1, max_length6, description验证码)3.生成token创建util.jwt_tuil.pyfrom 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在settings中修改同时记得修改.env.dev# 安全 secret_key: str dev-secret-key-123456-pydantic jwt_token_secret_key: str dhsjjdkfjdkfrjfrjgr-278783jkdsdhjdhjsds-dsdksjdkajskajieuiwueiwhdshmxzxno9iy token_expire_minutes: int 120 ALGORITHM:str HS256 cors_allow_origins: list[str] [*] REFRESH_TOKEN_EXPIRE_DAYS:int7SECRET_KEYdev-secret-key ALGORITHMHS256 JWT_TOKEN_SECRET_KEYdev-jwt-secret-key TOKEN_EXPIRE_MINUTES120 REFRESH_TOKEN_EXPIRE_DAYS7 CORS_ALLOW_ORIGINS[*]4、获取个人信息apisjob_seeker_router.get(/me, summary获取当前用户信息, description获取当前用户信息) async def get_me(job_seeker_idDepends(get_job_seeker_info)): logger.info(f通过token获取当前用户信息id:{job_seeker_id}) job_seeker await JobSeekerService.get_job_seeker_info_by_id(job_seeker_id) return { code: 1, message: 获取成功, data: job_seeker }servicestaticmethod async def get_job_seeker_info_by_id(job_seeker_id): job_seeker await JobSeeker.get_or_none(idjob_seeker_id) job_seeker.nickname f用户{job_seeker.mobile} if job_seeker.nickname is None else job_seeker.nickname job_seeker.avatar https://ts4.tc.mm.bing.net/th/id/OIP-C.kUV8yFYeWXeppllGjVqMrgAAAA?r0rs1pidImgDetMaino7rm3 return job_seeker5、注入依赖依赖注入Dependency Injection是一种设计思想函数 / 接口需要的外部资源数据库连接、用户认证、配置、工具类、参数校验不用自己内部创建而是由 FastAPI 框架自动提供、注入进来。FastAPI 使用内置Depends()实现这套机制不用第三方库。核心好处代码解耦业务逻辑和资源创建分离复用逻辑认证、数据库、分页等逻辑一处定义多处使用自动缓存、生命周期管理请求结束自动释放资源易于单元测试依赖可以轻松 Mock 替换上面代码中的参数就是注入依赖async def get_me(job_seeker_idDepends(get_job_seeker_info)):创建core.dependsfrom fastapi import Header from app.utils.jwt_util import verify_token def get_job_seeker_info(authorization: str|None Header(None)) - str: if not authorization: raise Exception(请登录) token authorization.replace(Bearer , ) payload verify_token(token) if payload is None: raise Exception(验证失败) job_seeker_id payload[user_id] return job_seeker_id在这里使用注入依赖的好处时不用在参数中传递token而是可以直接根据token在注入依赖时自动获取到当前登录的求职者id6、上传头像apisjob_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: 上传头像成功, }servicestaticmethod async def update_avatar(job_seeker_id, file): file_content await file.read() filename file.filename oss AliyunOSSTool() is_success, success_res oss.upload_single_file( file_content, filename, oss_pathjob_seeker_avatar/ ) if is_success: logger.info(上传成功) jobseeker await JobSeeker.get_or_none(idjob_seeker_id) jobseeker.avatarsuccess_res[access_url] await jobseeker.save() return { code: 1, message: 上传成功, data: success_res } else: raise Exception(上传失败)这里的头像上传借用了阿里云的对象存储oss配置流程先访问阿里云官方的oss管理控制台左侧导航栏最下面在sdk中访问OSS Python SDK V2打开官方配置文档向下滑点击RAM控制台超链接然后自行注册RAM用户填写表单后勾选使用永久accesskey创建成功后一定要复制给出的两个加密字符串然后回到控制台首页点击Bucket 列表新建一个bucket列表复制框选部分新建完成过后复制你的bucket名称然后回到项目中创建utils.oss_uti.py# oss_tool.py import oss2 import os from uuid import uuid4 from typing import List, Optional, Tuple from datetime import timedelta from oss2.exceptions import OssError from app.core.logging import logger from app.config.settings import settings class AliyunOSSTool: pip install oss22.19.1 pip install python-multipart 阿里云OSS工具类基于oss2封装封装OSS的文件上传、下载、删除、签名URL生成等核心操作 def __init__( self, access_key_id: Optional[str] None, access_key_secret: Optional[str] None, endpoint: Optional[str] None, bucket_name: Optional[str] None ): 初始化OSS工具类 :param access_key_id: 阿里云AccessKey ID :param access_key_secret: 阿里云AccessKey Secret :param endpoint: OSS地域节点如oss-cn-beijing.aliyuncs.com :param bucket_name: OSS Bucket名称说明若参数为None则从环境变量中读取对应配置 # 1. 加载配置优先传入参数其次环境变量 self.access_key_id access_key_id or settings.ALIYUN_OSS_ACCESS_KEY_ID self.access_key_secret access_key_secret or settings.ALIYUN_OSS_ACCESS_KEY_SECRET self.endpoint endpoint or settings.ALIYUN_OSS_ENDPOINT self.bucket_name bucket_name or settings.ALIYUN_OSS_BUCKET_NAME # 2. 校验配置完整性 self._check_config() # 3. 初始化OSS Auth和Bucket对象 self.auth oss2.Auth(self.access_key_id, self.access_key_secret) self.bucket oss2.Bucket( self.auth, self.endpoint, self.bucket_name, connect_timeout30, ) def _check_config(self) - None: 校验配置是否完整缺失则抛出异常 logger.info(OSS配置校验开始) logger.info(fAccessKey ID: {self.access_key_id}) logger.info(fAccessKey Secret: {* * min(len(self.access_key_secret or ), 8)}...) logger.info(fEndpoint: {self.endpoint}) logger.info(fBucket Name: {self.bucket_name}) # 检查配置是否为空或占位符 def is_valid_config(value: Optional[str]) - bool: 检查配置值是否有效非空且不是占位符 if not value: return False # 检查是否是占位符 placeholder_values [KEY, SECRET, your-, your_] return not any(placeholder in str(value) for placeholder in placeholder_values) required_configs { AccessKey ID: self.access_key_id, AccessKey Secret: self.access_key_secret, Endpoint: self.endpoint, Bucket Name: self.bucket_name } missing_configs [name for name, value in required_configs.items() if not is_valid_config(value)] if missing_configs: raise ValueError(fOSS配置缺失或无效{, .join(missing_configs)}请检查.env.dev文件或环境变量) def _generate_unique_filename(self, filename: str, oss_path: str) - str: 生成唯一的OSS文件路径避免文件名重复覆盖 :param filename: 原始文件名 :param oss_path: OSS存储目录 :return: 唯一的oss_file_key如uploads/abc123-test.jpg # 提取文件后缀 file_suffix os.path.splitext(filename)[1] # 生成UUID前缀保证唯一性 uuid_prefix str(uuid4())[:8] # 构造唯一文件名 unique_filename f{uuid_prefix}-{os.path.splitext(filename)[0]}{file_suffix} # 构造完整OSS文件路径 return f{oss_path.rstrip(/)}/{unique_filename} def upload_single_file(self, file_content: bytes, filename: str, oss_path: str uploads/, unique_filename: bool True) - Tuple[bool, dict]: 上传单个文件到OSS :param file_content: 文件字节内容 :param filename: 原始文件名 :param oss_path: OSS存储目录默认uploads/ :param unique_filename: 是否生成唯一文件名默认True避免覆盖 :return: (上传是否成功, 结果数据字典) try: # 1. 构造OSS文件路径 if unique_filename: oss_file_key self._generate_unique_filename(filename, oss_path) else: oss_file_key f{oss_path.rstrip(/)}/{filename} # 2. 调用oss2上传文件 result self.bucket.put_object(oss_file_key, file_content) # 3. 构造返回结果 if result.status 200: file_access_url fhttps://{self.bucket_name}.{self.endpoint}/{oss_file_key} return True, { filename: filename, oss_file_key: oss_file_key, access_url: file_access_url, file_size: len(file_content), etag: result.etag } else: return False, {error: fOSS返回非200状态码{result.status}} except OssError as e: return False, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, {error: f文件上传失败{str(e)}} def upload_multiple_files(self, files_data: List[Tuple[bytes, str]], oss_path: str uploads/, unique_filename: bool True) - List[dict]: 批量上传文件到OSS :param files_data: 文件数据列表每个元素为(文件字节内容, 原始文件名) :param oss_path: OSS存储目录默认uploads/ :param unique_filename: 是否生成唯一文件名默认True :return: 批量上传结果列表 upload_results [] for file_content, filename in files_data: success, result_data self.upload_single_file(file_content, filename, oss_path, unique_filename) upload_results.append({ filename: filename, status: success if success else failed, data: result_data }) return upload_results def download_file(self, oss_file_key: str) - Tuple[bool, Optional[oss2.models.GetObjectResult], Optional[dict]]: 从OSS下载文件返回文件流对象 :param oss_file_key: OSS文件完整路径如uploads/abc123-test.jpg :return: (下载是否成功, 文件流对象, 错误信息字典) try: # 1. 检查文件是否存在 if not self.bucket.object_exists(oss_file_key): return False, None, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 读取文件流oss2流式返回避免大文件内存溢出 file_object self.bucket.get_object(oss_file_key) return True, file_object, None except OssError as e: return False, None, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, None, {error: f文件下载失败{str(e)}} def delete_file(self, oss_file_key: str) - Tuple[bool, dict]: 从OSS删除指定文件 :param oss_file_key: OSS文件完整路径 :return: (删除是否成功, 结果数据字典) try: # 1. 检查文件是否存在 if not self.bucket.object_exists(oss_file_key): return False, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 调用oss2删除文件 result self.bucket.delete_object(oss_file_key) # 3. 验证删除结果 if result.status 204: return True, {oss_file_key: oss_file_key, message: 文件删除成功} else: return False, {error: fOSS返回非204删除状态码{result.status}} except OssError as e: return False, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, {error: f文件删除失败{str(e)}} def generate_sign_url(self, oss_file_key: str, expire_seconds: int 3600, method: str GET) - Tuple[ bool, Optional[str], dict]: 生成OSS文件临时签名URL私有Bucket必备用于临时访问/下载 :param oss_file_key: OSS文件完整路径 :param expire_seconds: URL有效期默认3600秒/1小时 :param method: 访问方法GET下载/访问PUT上传 :return: (生成是否成功, 签名URL, 结果数据字典) try: # 1. 检查文件是否存在可选根据业务需求决定是否开启 if not self.bucket.object_exists(oss_file_key): return False, None, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 调用oss2生成签名URL sign_url oss2.sign_url( self.auth, self.endpoint, self.bucket_name, oss_file_key, expire_seconds, methodmethod ) return True, sign_url, {expire_seconds: expire_seconds, method: method} except OssError as e: return False, None, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, None, {error: f签名URL生成失败{str(e)}}然后配置settings和.env.dev这四行换成刚才复制的四个部分自己使用自己的配置配置流程就完成了以后图片就会上传到阿里云的远程对象存储oss中

模拟BOSS招聘项目——求职者端功能实现

相关推荐

游戏开发中的定时器与冷却机制实现

户外露营净水器怎么选？从过滤原理到实战场景的完整技术指南

Boss招聘平台--求职者登录

更多资讯

aiohttp 修复 `CookieJar.update_cookies()` 未复制用户传入的可变 `Morsel` 对象的缺陷解析

OpenSearch查询DSL完全指南：Bool、Term、Range、Wildcard等10大查询类型一篇讲透

Gyroflow 开源视频稳定工具使用指南：用陀螺仪数据消除画面抖动

免费 3 步下载流媒体：DASH/HLS 课程与直播的本地保存方法

5个sina邮箱开发避坑点：新手速查手册

公租房摇号时间源码深度剖析：3个技巧搞定性能优化

今日更新

OneUptime 自定义探针（Custom Probe）部署实战：私网监控、代理配置与断连排障全指南

大众TL52625前端框架材料要求详解：从性能测试到落地执行

TiXL 浮点运算算子库 Lib.numbers.float 完全指南：44 个算子的参数详解、源码原理与实战串联

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？