FastAPI 实战 day07：企业认证与附件简历功能开发

1. 企业认证接口：事务 + 多文件上传

企业认证是今天新增的完整功能，它最能体现"真实业务"的复杂度：要同时写 3 张表、传 3 个文件（营业执照 + 法人身份证正反面）。

1.1 接口层：接收表单 + 三个文件

# app/apis/enterprise_api.py

@enterprise_api.post("/save", summary="保存企业信息")
async def saveEnterpriseInfo(
form: EnterpriseCreateRequest = Depends(get_enterprise_form), # 表单字段自动解析
business_license_file: UploadFile = File(None, description="营业执照"),
legal_id_front_file: UploadFile = File(None, description="法人身份证（正面）"),
legal_id_back_file: UploadFile = File(None, description="法人身份证（反面）"),
):
await EnterpriseService.saveEnterpriseInfo(
form, business_license_file, legal_id_front_file, legal_id_back_file
)
return {"code": 1, "message": "保存成功"}

注意 File(None) 里的 None：表示文件"可以不传"，但一旦传了就是 UploadFile 对象。

1.2 依赖层：用 Form 把一堆表单字段打包成一个对象

企业信息字段特别多，如果接口参数一个个写会很丑。我们用 get_enterprise_form 把它们收拢成一个 EnterpriseCreateRequest：

# app/core/depends.py

async def get_enterprise_form(
enterprise_name=Form(..., description="企业名称"),
unified_social_credit_code=Form(..., description="统一社会信用代码"),
legal_representative=Form(..., description="法人代表"),
registered_capital=Form(..., description="注册资本"),
establish_date=Form(None, description="成立日期"), # None 表示可空
register_status=Form(..., description="登记状态"),
industry_id=Form(..., description="所属行业"),
company_scale=Form(..., description="公司规模"),
headquarters_address=Form(..., description="总部地址"),
business_scope=Form(..., description="经营范围"),
contact_name=Form(..., description="联系人姓名"),
contact_phone=Form(..., description="联系电话"),
contact_email=Form(..., description="联系邮箱"),
# … 其他字段
):
return EnterpriseCreateRequest(
enterprise_name=enterprise_name,
unified_social_credit_code=unified_social_credit_code,
# … 一一对应
)

Form(…) 里的 … 表示必填，Form(None) 表示选填。

1.3 业务层：事务保证"要么全成，要么全不成"

企业认证要连写 3 张表、传 3 个文件。最怕写到第 2 张表时挂了，第 1 张却已经写了——数据就脏了。

事务（Transaction） 就是用来解决这个问题的：事务里的所有操作，要么全部成功提交，要么整体回滚。

# app/services/enterprise_services.py

import uuid
from fastapi import UploadFile
from tortoise.transactions import in_transaction
from app.utils.oss_util import AliyunOSSTool

class EnterpriseService:
@staticmethod
async def saveEnterpriseInfo(form, business_license_file, legal_id_front_file, legal_id_back_file):
async with in_transaction() as conn: # 开启事务
# 1) 主表：企业账号（状态=待审核）
enterprise = await Enterprise.create(
enterprise_name=form.enterprise_name,
enterprise_code=str(uuid.uuid4()), # 生成唯一企业码
account_status=AccountStatus.PENDING_AUDIT, # 待审核
blacklist_status=BlackListStatus.NOT_BANNED,
audit_type=AuditType.NEW_ENTERPRISE_AUTH,
submit_time=now(),
)

# 2) 子表：企业详细信息
await EnterpriseInfo.create(
enterprise_name=form.enterprise_name,
unified_social_credit_code=form.unified_social_credit_code,
# … 其它字段
enterprise_id=enterprise.id
)

# 3) 上传三个文件到阿里云 OSS，拿到可访问的 URL
oss = AliyunOSSTool()
lic = await business_license_file.read()
ok1, res1 = oss.upload_single_file(lic, business_license_file.filename, oss_path="enterprise/")
if not ok1:
raise Exception("上传营业执照图片失败") # 抛异常 → 事务自动回滚

front = await legal_id_front_file.read()
ok2, res2 = oss.upload_single_file(front, legal_id_front_file.filename, oss_path="enterprise/")
if not ok2:
raise Exception("上传法人身份证正面图片失败")

back = await legal_id_back_file.read()
ok3, res3 = oss.upload_single_file(back, legal_id_back_file.filename, oss_path="enterprise/")
if not ok3:
raise Exception("上传法人身份证反面图片失败")

# 4) 子表：资质信息（存文件 URL）
await EnterpriseQualification.create(
contact_name=form.contact_name,
contact_phone=form.contact_phone,
enterprise_id=enterprise.id,
business_license_url=res1["access_url"],
legal_id_front_url=res2["access_url"],
legal_id_back_url=res3["access_url"]
)
# 离开 with 代码块 = 事务提交成功

事务就像一个打包操作。我们用 async with in_transaction() 包住所有写库动作。只要中间任何一步抛异常（比如 OSS 上传失败），前面已经写入的数据会自动撤销，不会出现半成品。

OSS（对象存储） 可以理解成"云上的硬盘"。文件不直接存服务器，而是上传到阿里云，拿到一个 access_url 存进数据库，前端用这个 URL 就能展示图片。

2. 附件简历：OSS 上传 + 数量限制 + 默认简历

求职者可以上传多个附件简历（Word/PDF），但只允许最多 3 个，且其中只能有 1 个是"默认简历"。来看看怎么实现这两个约束。

2.1 接口层：保存 / 更新 / 删除

# app/apis/resume_api.py

@resume_api.post("/upload", summary="上传附件简历")
async def upload_resume(
user_id: int = Form(...),
resume_file: UploadFile = File(..., description="简历文件（Word/PDF）"),
is_default: bool = Form(False, description="是否设为默认简历"),
):
result = await ResumeService.upload_resume(user_id, resume_file, is_default)
return {"code": 1, "message": "上传成功", "data": result}

@resume_api.put("/set-default", summary="设置默认简历")
async def set_default_resume(
resume_id: int = Form(...),
user_id: int = Form(...),
):
await ResumeService.set_default_resume(resume_id, user_id)
return {"code": 1, "message": "设置成功"}

@resume_api.delete("/delete", summary="删除附件简历")
async def delete_resume(
resume_id: int = Form(...),
user_id: int = Form(...),
):
await ResumeService.delete_resume(resume_id, user_id)
return {"code": 1, "message": "删除成功"}

2.2 业务层：校验数量与默认约束

# app/services/resume_services.py

from tortoise.transactions import in_transaction

class ResumeService:
@staticmethod
async def upload_resume(user_id: int, resume_file: UploadFile, is_default: bool):
# 1) 校验数量：最多 3 个
count = await Resume.filter(user_id=user_id).count()
if count >= 3:
raise HTTPException(status_code=400, detail="附件简历最多只能上传 3 个")

# 2) 如果设为默认，先把其他简历的默认标记取消
if is_default:
await Resume.filter(user_id=user_id, is_default=True).update(is_default=False)

# 3) 上传到 OSS
oss = AliyunOSSTool()
content = await resume_file.read()
ok, result = oss.upload_single_file(content, resume_file.filename, oss_path="resumes/")
if not ok:
raise HTTPException(status_code=500, detail="简历文件上传失败")

# 4) 保存记录
resume = await Resume.create(
user_id=user_id,
file_url=result["access_url"],
file_name=resume_file.filename,
is_default=is_default,
)
return {"resume_id": resume.id, "file_url": result["access_url"]}

@staticmethod
async def set_default_resume(resume_id: int, user_id: int):
async with in_transaction() as conn:
# 先把该用户所有简历的默认标记取消
await Resume.filter(user_id=user_id).update(is_default=False)
# 再把目标简历设为默认
await Resume.filter(id=resume_id, user_id=user_id).update(is_default=True)

@staticmethod
async def delete_resume(resume_id: int, user_id: int):
resume = await Resume.get_or_none(id=resume_id, user_id=user_id)
if not resume:
raise HTTPException(status_code=404, detail="简历不存在")

# 删除 OSS 文件（可选）
oss = AliyunOSSTool()
oss.delete_file(resume.file_url)

# 删除数据库记录
await resume.delete()

3. 总结

今天我们完成了两个核心业务功能：

事务和 OSS 是后端开发中的"基础设施"技能。掌握了它们，你就能应对绝大多数需要"写多张表 + 存文件"的真实业务场景。