FastAPI + Tortoise-ORM + 阿里云 OSS 实战：职位投递链路与招聘团队模块的设计

FastAPI + Tortoise-ORM + 阿里云 OSS 实战：职位投递链路与招聘团队模块的设计

一、前言介绍

1.1 背景与功能定位

招聘平台走到"职位"之后，真正的业务闭环是求职者把简历投给某个职位，招聘方在后台收到并筛选。这一段要解决两件事：其一，招聘方不是只有一个"企业账号"，而是有若干个招聘团队成员（HR、用人经理），他们各自发布职位、查看收到的简历；其二，求职者点击"投递"，系统要把他的附件简历复制一份快照存到投递记录里，避免原简历被改后投递记录跟着失真。

1.2 数据模型总览

Enterprise（企业主表） └── 1 : N ── RecruitTeam（招聘团队，member 账号，软删除） └── recruit_team_id ── Job（职位，由团队成员发布） JobSeeker（求职者） └── 1 : N ── AttachmentResume（附件简历，含 file_url） └── job_seeker_id ── ResumeSubmissionRecords（投递记录） ├── job_id → Job └── job_seeker_id → JobSeeker

要点：职位归属到团队而非企业，投递记录用job_id + job_seeker_id双列定位一次投递，附件简历通过job_seeker_id关联，但投递时只"复制"其 URL，不建立强外键。

1.3 投递流程总览

求职者（持 JWT）POST /job/resume_submission?job_id=xxx → 取该求职者"默认/已删除"附件简历的 file_url → 用雪花 ID 生成新文件名，OSS 同桶拷贝出一份副本 → 写 ResumeSubmissionRecords（状态=已投递） 招聘方（持团队 JWT）GET /enterprise/resume_center → 查该团队发布的所有职位 → 每个职位下的投递记录 → 关联求职者 + 求职意向 + 工作经历 → 拼成简历中心列表

二、环境准备

2.1 依赖与工具

投递链路依赖两个工具类，与之前模块同源：

• app.utils.oss_util.AliyunOSSTool：阿里云 OSS 封装，本次用到同桶拷贝copy_single_file；
• app.utils.snowflake_util.SnowflakeSingleton：雪花算法单例，生成全局唯一文件 ID；
• app.models.job.ResumeStatus（IntEnum）：投递状态枚举。

2.2 路由与鉴权

• 求职者投递：/job/resume_submission，鉴权依赖get_job_info（从 JWT 解析出job_seeker_id）；
• 招聘方简历中心：/enterprise/resume_center，鉴权依赖同一个get_job_info（JWT 里user_id实际装的是团队 ID）；
• 团队登录：/enterprise/team_login，手机号 + 短信验证码，通过后签发双 Token。

2.3 模型注册

app/models/__init__.py中新增导出：

fromapp.models.enterpriseimportEnterprise,EnterpriseInfo,EnterpriseQualification,EnterpriseReview,RecruitTeamfromapp.models.jobimportJob,ResumeSubmissionRecords

• 第 1 行：把RecruitTeam纳入 ORM 注册，迁移才认得这张表；
• 第 2 行：ResumeSubmissionRecords同理；
• 漏掉这一步会触发Model not registered之类的启动报错，是新模型最容易被遗忘的一处。

三、知识点讲解

3.1 招聘团队为什么要独立成表

企业账号和"发布职位的 HR"是两回事：一个企业要多个 HR 协作，HR 离职要能禁用而不影响企业本身。所以把RecruitTeam单独建表，用enterprise_id整型关联企业，并带status（正常/禁用）与is_deleted（软删除）。

classRecruitTeam(Model):mobile=fields.CharField(max_length=20,description="手机号")password=fields.CharField(max_length=256,description="登录密码（加密存储）",null=True)status=fields.IntEnumField(enum_type=TeamMemberStatus,description="状态：1正常 2禁用")enterprise_id=fields.IntField(null=True,description="企业ID")is_deleted=fields.IntEnumField(enum_type=DeleteStatus,default=DeleteStatus.NOT_DELETED,...)

• status用枚举：禁用后该成员无法登录，职位仍归属企业，不丢失；
• is_deleted软删除：不真删行，列表查询加is_deleted=NOT_DELETED过滤即可"隐身"。

3.2 投递记录为什么复制简历而非引用

附件简历是会被用户替换、删除的。如果投递记录只存attachment_resume_id，用户改了原简历，历史投递记录也跟着变，招聘方看到的就不是"当时投的那份"。所以投递时把简历文件在 OSS 上拷贝一份副本，记录副本 URL：

copy_resume_url=fields.CharField(max_length=256,description="附件简历URL")

• 字段只存 URL 字符串，不建外键，断开了与原简历的生命周期耦合；
• 即便原简历被删，投递记录里的副本 URL 依然有效。

3.3 OSS 同桶拷贝的注意点

复制不是把文件下载再上传，而是调用 OSS 的copy_object，服务端内部搬数据，节省带宽也更快：

result=self.bucket.copy_object(src_bucket,source_oss_key,target_oss_key)

• src_bucket与目标桶是同一个（同桶拷贝），直接传当前bucket_name；
• 目标已存在且overwrite=False会拒绝覆盖，避免误冲掉别人已投的副本。

3.4 雪花 ID 做文件名的意义

投递副本的文件名如果用uuid4每次都不同没问题，但这里选了雪花算法，除了唯一还带时间有序：

snowflake=SnowflakeSingleton.get_instance(worker_id=1)file_id=snowflake.get_id()

• 单例保证全进程只有一个生成器，worker_id首次必须传入、之后锁定；
• 生成的 ID 是 64 位整数，毫秒级有序，便于按文件名粗略排序、排查问题。

四、代码逻辑拆解

4.1 招聘团队登录 team_login

team=awaitRecruitTeam.filter(mobile=login.mobile,status=TeamMemberStatus.NORMAL,is_deleted=DeleteStatus.NOT_DELETED).first()ifnotteam:raiseException("手机号不存在或账号存在异常")key=f"boss-api:enterprise-login:sms:{login.mobile}"redis_code=redis_client.get(key)ifredis_codeisNone:raiseException("验证码已过期")ifredis_code!=login.code:raiseException("验证码错误")access_token,refresh_token=create_tokens(str(team.id),login.mobile)redis_client.delete(key)

• 第 1–3 行：按手机号查成员，同时要求"状态正常 + 未软删除"，被禁用或已删的账号直接查不到；
• 第 4–8 行：验证码从 Redis 取，过期/错误分别报错，流程与企业登录一致；
• create_tokens(str(team.id), ...)：user_id装的是团队 ID，所以后续get_job_info解析出来的就是团队成员身份，能查到他发布的职位与收到的简历；
• 最后一行：验证码用后即焚，防重放。

4.2 审核通过时自动建招聘成员

企业审核通过的副作用是给联系人自动生成一个招聘团队成员账号：

ent=awaitEnterpriseQualification.filter(enterprise_id=enterprise_id).first()awaitRecruitTeam.create(name=ent.contact_name,mobile=ent.contact_phone,email=ent.contact_email,enterprise_id=enterprise_id,status=TeamMemberStatus.NORMAL,)

• 用资质表里的联系人信息建号，省去企业方再单独录入 HR；
• status=NORMAL：建好即可登录，不需要二次激活。

4.3 职位投递核心 resume_submission

这是整条链路的核心，逐段看：

oss=AliyunOSSTool()att=awaitAttachmentResume.filter(job_seeker_id=job_seeker_id,is_deleted=True).first()file_url=att.file_url source_key=file_url.replace(f"https://{oss.bucket_name}.{oss.endpoint}/","")

• 第 1 行：初始化 OSS 工具；
• 第 2 行：取该求职者的附件简历。注意这里过滤条件是is_deleted=True——与原简历表"未删除=0"的约定相反，是个易踩的坑（见问题排查 5.1）；
• 第 3–4 行：从完整 URL 里把 OSS key 抠出来：把https://桶名.域名/前缀替换掉，剩下的就是对象路径，拷贝接口只认 key。

snowflake=SnowflakeSingleton.get_instance(worker_id=1)file_id=snowflake.get_id()source_filename=os.path.basename(source_key)target_key=f"job_seeker_avatar/copy/{file_id}-{source_filename}"copy_ok,copy_data=oss.copy_single_file(source_oss_key=source_key,target_oss_key=target_key,overwrite=False)

• 第 1–2 行：拿雪花 ID 当文件前缀，保证副本名全局唯一；
• os.path.basename(source_key)：只取文件名，避免把原目录结构拼进新路径造成重复嵌套；
• target_key落在job_seeker_avatar/copy/下，和原简历目录分开，语义清晰；
• overwrite=False：同名不覆盖，配合唯一前缀基本不会撞。

awaitResumeSubmissionRecords.create(job_id=job_id,job_seeker_id=job_seeker_id,copy_resume_url=copy_data["target_url"],submit_time=now(),resume_status=ResumeStatus.SUBMITTED)

• 落库四要素：job_id（投给哪个职位）、job_seeker_id（谁投的）、copy_resume_url（副本地址）、resume_status（初始"已投递"）；
• submit_time=now()：写入投递时间，后续简历中心按它排序展示。

4.4 招聘方简历中心 resume_center

jobs=awaitJob.filter(recruit_team_id=team_id)data_dict_list=[]foriinjobs:resume_submission_records=awaitResumeSubmissionRecords.filter(job_id=i.id)forjinresume_submission_records:job_seeker=awaitJobSeeker.filter(id=j.job_seeker_id).first()jobIntention=awaitJobIntention.filter(job_seeker_id=j.job_seeker_id).first()work=awaitWorkExperience.filter(job_seeker_id=j.job_seeker_id).order_by("-entry_time")info_dict={"job_seeker":job_seeker,"jobIntention":jobIntention,"workExperience":work,"resume_status":j.resume_status,"submit_time":j.submit_time,}data_dict_list.append(info_dict)

• 第 1 行：先取"我（团队）发布的所有职位"；
• 内层循环：每个职位下的投递记录，再逐条关联求职者、求职意向、工作经历；
• order_by("-entry_time")：工作经历按入职时间倒序，最新一段排前面；
• 每条投递拼成字典，最终是"职位 → 投递者 → 简历内容"的扁平列表。

4.5 投递接口签名

@job_router.post("/resume_submission",summary="职位简历提交")asyncdefresume_submission(id=Depends(get_job_info),job_id:int=Query(...,description="职位简历")):awaitJobService.resume_submission(id,job_id)return{"code":1,"message":"职位简历提交成功"}

• id=Depends(get_job_info)：从 JWT 拿到求职者 ID，前端无法伪造"替别人投递"；
• job_id=Query(...)：...表示必填，不传直接 422；
• 投递动作本身无请求体，参数走查询字符串即可。