本文记录一个 Boss 直聘类招聘平台中「企业管理端」后端的真实实现企业提交认证材料、管理员审核、企业验证码登录全流程。重点分享接口设计思路以及开发过程中踩到的 6 个 Tortoise / FastAPI 实战深坑。一、功能与技术栈企业认证是招聘平台的信任基石。本模块实现以下能力企业端提交营业执照、法人身份证等材料完成认证管理端分页查看待审企业、查看详情、通过 / 拒绝审核企业登录手机号 短信验证码登录需审核通过技术栈层选型Web 框架FastAPIORMTortoise ORM异步数据库MySQL / PostgreSQL缓存Redis存验证码对象存储阿里云 OSS资质材料鉴权JWTaccess refresh token接口统一约定{code: 1, message: ..., data: ...}表示成功code0表示失败。二、数据模型设计核心涉及 5 张表字段用IntEnum约束状态值避免脏数据。from enum import IntEnum from tortoise import fields, models ​ class AccountStatus(IntEnum): NORMAL 0 # 正常 PENDING_AUDIT 1 # 待审核 BANNED 2 # 封禁 ​ class AuditType(IntEnum): NEW_ENTERPRISE_AUTH 1 # 新企业认证 QUALITY_UPDATE 2 # 资质更新 INFO_CHANGE 3 # 信息变更 ​ class Enterprise(models.Model): 企业主表 id fields.IntField(pkTrue) enterprise_name fields.CharField(max_length255) enterprise_code fields.CharField(max_length100, uniqueTrue) account_status fields.IntEnumField(enum_typeAccountStatus, description账号状态0正常,1待审核,2封禁) audit_type fields.IntEnumField(enum_typeAuditType) submit_time fields.DatetimeField(nullTrue) ​ class EnterpriseInfo(models.Model): 企业信息表营业执照信息等 id fields.IntField(pkTrue) enterprise_name fields.CharField(max_length255) unified_social_credit_code fields.CharField(max_length50, uniqueTrue) legal_representative fields.CharField(max_length100) register_status fields.IntEnumField(enum_typeRegisterStatus) industry fields.ForeignKeyField(models.IndustryPosition, related_nameenterprise_info_list) enterprise_id fields.IntField(nullTrue) ​ class EnterpriseQualification(models.Model): 资质材料表 id fields.IntField(pkTrue) contact_phone fields.CharField(max_length32) business_license_url fields.CharField(max_length512, nullTrue) legal_id_front_url fields.CharField(max_length512, nullTrue) legal_id_back_url fields.CharField(max_length512, nullTrue) enterprise_id fields.IntField(nullTrue) ​ class EnterpriseReview(models.Model): 审核主表一对一 id fields.IntField(pkTrue) review_result fields.IntField(nullTrue) # 1通过 2驳回 review_reason fields.CharField(max_length512, nullTrue) audit_time fields.DatetimeField(nullTrue) enterprise_id fields.IntField(nullTrue) ​ class EnterpriseReviewRecord(models.Model): 审核流水表一对多记录每次操作 id fields.IntField(pkTrue) review_result fields.IntField(nullTrue) review_reason fields.CharField(max_length512, nullTrue) create_time fields.DatetimeField(auto_now_addTrue) enterprise_id fields.IntField()三、接口一览方法路径说明POST/enterprise/save企业提交认证材料GET/enterprise/list管理端分页查询企业列表GET/enterprise/detail/{id}查询企业认证详情POST/enterprise/review管理员审核通过/拒绝POST/enterprise/send_sms_code发送登录验证码POST/enterprise/login手机号 验证码登录四、核心实现讲解4.1 企业提交认证事务 文件上传提交涉及创建企业主表 信息表 资质表 流水表且有 3 个文件要上传 OSS必须保证原子性否则会出现「主表创建成功但文件上传失败」的脏数据。staticmethod async def saveEnterpriseInfo(form, business_license_file, legal_id_front_file, legal_id_back_file): # 校验必传文件 if not business_license_file: raise Exception(请上传营业执照) if not legal_id_front_file: raise Exception(请上传法人身份证正面) if not legal_id_back_file: raise Exception(请上传法人身份证反面) ​ async with in_transaction() as conn: enterprise await Enterprise.create( enterprise_nameform.enterprise_name, enterprise_codestr(uuid.uuid4()), account_statusAccountStatus.PENDING_AUDIT, # 新建即待审核 blacklist_statusBlackListStatus.NOT_BANNED, audit_typeAuditType.NEW_ENTERPRISE_AUTH, submit_timenow(), ) await EnterpriseInfo.create( enterprise_nameform.enterprise_name, unified_social_credit_codeform.unified_social_credit_code, legal_representativeform.legal_representative, industry_idform.industry_id, enterprise_identerprise.id, ) # 上传三个文件到 OSS oss AliyunOSSTool() is_ok, res oss.upload_single_file( await business_license_file.read(), business_license_file.filename, oss_pathenterprise/) if not is_ok: raise Exception(上传营业执照图片失败) # ... 其余两张同理 await EnterpriseQualification.create( contact_phoneform.contact_phone, business_license_urlres[access_url], enterprise_identerprise.id, ) await EnterpriseReviewRecord.create( remark提交认证申请, enterprise_identerprise.id)in_transaction()保证任一文件上传失败抛出异常前面创建的记录全部回滚。4.2 审核接口状态同步 防重复这是业务核心也是最容易出 bug 的地方。审核要同时做三件事写 / 更新审核主表追加一条审核流水同步回写企业账号状态并且必须防止重复审核。staticmethod async def enterprise_review(req: EnterpriseReviewCreateRequest): enterprise_id req.enterprise_id enterprise await Enterprise.get_or_none(identerprise_id) if enterprise is None: raise Exception(企业不存在) ​ # ① 防重复审核已有最终结果则拒绝 enterprise_review await EnterpriseReview.get_or_none(enterprise_identerprise_id) if enterprise_review is not None and enterprise_review.review_result is not None: raise Exception(该企业已审核请勿重复审核) ​ # ② 参数校验 review_result req.review_result if review_result not in (1, 2): raise Exception(审核结果参数错误) ​ # ③ 创建或更新审核主表 if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultreview_result, review_reasonreq.review_reason, audit_timenow()) else: enterprise_review.review_result review_result enterprise_review.review_reason req.review_reason enterprise_review.audit_time now() await enterprise_review.save() ​ # ④ 追加流水 await EnterpriseReviewRecord.create( enterprise_identerprise_id, remark审核通过 if review_result 1 else 审核未通过, review_resultreview_result, review_reasonreq.review_reason) ​ # ⑤ 同步账号状态通过→正常拒绝→封禁 enterprise.account_status ( AccountStatus.NORMAL if review_result 1 else AccountStatus.BANNED) await enterprise.save()关键点审核「拒绝」时也要回写account_status BANNED。很多同学只处理了通过的逻辑导致拒绝后企业账号仍显示正常——这就是典型的状态机不同步 bug。4.3 验证码登录登录前先发码存 Redis带过期登录时校验staticmethod async def send_sms_code(mobile: str): code get_random_code() # 6 位随机 redis_client.setex(fboss-api:enterprise-login:sms:{mobile}, 300, code) ​ staticmethod async def login(req: LoginMobileRequest): quals await EnterpriseQualification.filter(contact_phonereq.mobile) if not quals: # 注意filter 返回列表用 not 判断 raise Exception(手机号不存在) for q in quals: review await EnterpriseReview.get_or_none(enterprise_idq.enterprise_id) if review is None: continue # 无审核记录跳过 if review.review_result 1: # 仅审核通过可登录 redis_code redis_client.get(fboss-api:enterprise-login:sms:{req.mobile}) if redis_code is None: raise Exception(验证码已过期) if redis_code ! req.code: raise Exception(验证码错误) access, refresh create_tokens(str(q.enterprise_id), req.mobile) redis_client.delete(fboss-api:enterprise-login:sms:{req.mobile}) return {enterprise_access_token: access, enterprise_refresh_token: refresh} raise Exception(账号未审核通过)五、实战踩坑记录重点下面这些是开发时真实踩过的坑每一个都会让接口直接 500。坑 1Tortoise 模型对象不能直接返回select_enterprise_list最初直接把模型对象塞进 dict# ❌ 崩溃FastAPI 无法把 Tortoise 模型 / 关联对象序列化成 JSON return {enterprise: enterprise, industry: enterpriseinfo.industry.name}# ✅ 手动拆字段 return { enterprise: {id: enterprise.id, enterprise_name: enterprise.enterprise_name, account_status: int(enterprise.account_status) ...}, industry: enterpriseinfo.industry.name if enterpriseinfo and enterpriseinfo.industry else None }坑 2IntEnumField 的「0 值」序列化雷account_status为0正常时下面这行会把 0 当成 False# ❌ 0 被认为是 False结果返回 None前端映射失败 account_status: int(enterprise.account_status) if enterprise.account_status else None # ✅ 显式判断 None account_status: int(enterprise.account_status) if enterprise.account_status is not None else NonePython 中0、、[]都是 falsy用if x判断枚举值极度危险。坑 3filter()返回的是查询集不是 None# ❌ 永远不为 None校验失效 if enterprise_qualifications is None: raise Exception(手机号不存在) # ✅ 用 not 判断空列表 if not enterprise_qualifications: raise Exception(手机号不存在)坑 4关联对象空指针enterpriseinfo可能不存在或未关联industry直接.industry.name必崩# ❌ industry: enterpriseinfo.industry.name # ✅ industry: enterpriseinfo.industry.name if enterpriseinfo and enterpriseinfo.industry else None坑 5字段名拼写不一致审核表字段是audit_time代码里误写成review_time一调用就报「未知字段」# ❌ EnterpriseReview.create(..., review_timenow()) # ✅ EnterpriseReview.create(..., audit_timenow())坑 6重复审核无防护最初审核接口对已审核企业通过/拒绝没有任何限制管理员可以无限次点「通过」产生大量冗余流水且状态混乱。务必在写入前判断review_result is not None见 4.2 的①。六、总结一个看似简单的「企业认证审核」功能背后藏着事务一致性、状态机同步、防重复提交、ORM 序列化与 Python falsy 判断等多个雷区。核心经验ORM 模型永远手动序列化再返回不要直接丢给 FastAPI。枚举值判断用is not None别用if x。filter()用not判空别用is None。每个写接口都要考虑状态机是否正确同步、是否允许重复操作。关联对象一律做空值保护。以上即为本模块完整实现欢迎在评论区交流 Tortoise ORM 的更多坑。