1. 企业认证接口：事务 + 多文件上传
企业认证是今天新增的完整功能，它最能体现"真实业务"的复杂度：要同时写 3 张表、传 3 个文件（营业执照 + 法人身份证正反面）。
1.1 接口层：接收表单 + 三个文件
# app/apis/enterprise_api.py@enterprise_api.post("/save",summary="保存企业信息")asyncdefsaveEnterpriseInfo(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="法人身份证（反面）"),):awaitEnterpriseService.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.pyasyncdefget_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="联系邮箱"),# ... 其他字段):returnEnterpriseCreateRequest(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.pyimportuuidfromfastapiimportUploadFilefromtortoise.transactionsimportin_transactionfromapp.utils.oss_utilimportAliyunOSSToolclassEnterpriseService:@staticmethodasyncdefsaveEnterpriseInfo(form,business_license_file,legal_id_front_file,legal_id_back_file):asyncwithin_transaction()asconn:# 开启事务# 1) 主表：企业账号（状态=待审核）enterprise=awaitEnterprise.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) 子表：企业详细信息awaitEnterpriseInfo.create(enterprise_name=form.enterprise_name,unified_social_credit_code=form.unified_social_credit_code,# ... 其它字段enterprise_id=enterprise.id)# 3) 上传三个文件到阿里云 OSS，拿到可访问的 URLoss=AliyunOSSTool()lic=awaitbusiness_license_file.read()ok1,res1=oss.upload_single_file(lic,business_license_file.filename,oss_path="enterprise/")ifnotok1:raiseException("上传营业执照图片失败")# 抛异常 → 事务自动回滚front=awaitlegal_id_front_file.read()ok2,res2=oss.upload_single_file(front,legal_id_front_file.filename,oss_path="enterprise/")ifnotok2:raiseException("上传法人身份证正面图片失败")back=awaitlegal_id_back_file.read()ok3,res3=oss.upload_single_file(back,legal_id_back_file.filename,oss_path="enterprise/")ifnotok3:raiseException("上传法人身份证反面图片失败")# 4) 子表：资质信息（存文件 URL）awaitEnterpriseQualification.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="上传附件简历")asyncdefupload_resume(user_id:int=Form(...),resume_file:UploadFile=File(...,description="简历文件（Word/PDF）"),is_default:bool=Form(False,description="是否设为默认简历"),):result=awaitResumeService.upload_resume(user_id,resume_file,is_default)return{"code":1,"message":"上传成功","data":result}@resume_api.put("/set-default",summary="设置默认简历")asyncdefset_default_resume(resume_id:int=Form(...),user_id:int=Form(...),):awaitResumeService.set_default_resume(resume_id,user_id)return{"code":1,"message":"设置成功"}@resume_api.delete("/delete",summary="删除附件简历")asyncdefdelete_resume(resume_id:int=Form(...),user_id:int=Form(...),):awaitResumeService.delete_resume(resume_id,user_id)return{"code":1,"message":"删除成功"}2.2 业务层：校验数量与默认约束
# app/services/resume_services.pyfromtortoise.transactionsimportin_transactionclassResumeService:@staticmethodasyncdefupload_resume(user_id:int,resume_file:UploadFile,is_default:bool):# 1) 校验数量：最多 3 个count=awaitResume.filter(user_id=user_id).count()ifcount>=3:raiseHTTPException(status_code=400,detail="附件简历最多只能上传 3 个")# 2) 如果设为默认，先把其他简历的默认标记取消ifis_default:awaitResume.filter(user_id=user_id,is_default=True).update(is_default=False)# 3) 上传到 OSSoss=AliyunOSSTool()content=awaitresume_file.read()ok,result=oss.upload_single_file(content,resume_file.filename,oss_path="resumes/")ifnotok:raiseHTTPException(status_code=500,detail="简历文件上传失败")# 4) 保存记录resume=awaitResume.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"]}@staticmethodasyncdefset_default_resume(resume_id:int,user_id:int):asyncwithin_transaction()asconn:# 先把该用户所有简历的默认标记取消awaitResume.filter(user_id=user_id).update(is_default=False)# 再把目标简历设为默认awaitResume.filter(id=resume_id,user_id=user_id).update(is_default=True)@staticmethodasyncdefdelete_resume(resume_id:int,user_id:int):resume=awaitResume.get_or_none(id=resume_id,user_id=user_id)ifnotresume:raiseHTTPException(status_code=404,detail="简历不存在")# 删除 OSS 文件（可选）oss=AliyunOSSTool()oss.delete_file(resume.file_url)# 删除数据库记录awaitresume.delete()3. 总结
今天我们完成了两个核心业务功能：
| 功能 | 技术要点 | 业务价值 |
|---|---|---|
| 企业认证 | 事务 + 多文件上传 + OSS | 保证数据一致性，避免脏数据 |
| 附件简历 | 数量校验 + 默认约束 + OSS 联动 | 满足产品约束，防止用户误操作 |
事务和 OSS 是后端开发中的"基础设施"技能。掌握了它们，你就能应对绝大多数需要"写多张表 + 存文件"的真实业务场景。
)