基于 FastAPI + Tortoise ORM 的企业端注册认证模块设计与实现

关键词：FastAPI、Tortoise ORM、企业认证、多表事务、阿里云 OSS、Pydantic v2

前言

在招聘类平台（如 BOSS 直聘类产品）中，企业端注册和普通用户注册有本质区别：企业不是"填个账号密码就能用"，而是必须提交营业执照、法人身份证、统一社会信用代码等资质材料，经过平台审核后才能发布职位。本文复盘我最近落地的一个企业端注册/认证后端模块，从业务场景、数据库设计到开发流程完整拆解，并总结几个实战中踩到的坑。

技术栈：Python 3.11 + FastAPI + Tortoise ORM（异步） + 阿里云 OSS + SQLite/MySQL。

一、业务场景分析

1.1 业务定位

企业端用户的核心诉求是"尽快通过审核、开始招人"，而平台的核心诉求是"确保入驻企业真实合法、可控可封禁"。两者天然存在张力，注册模块就要在这之间做平衡：

• 对企业：表单一次填完、材料可上传、提交后进入"待审核"状态；

• 对平台：多维度资质留存（营业执照、身份证、信用代码）、状态机管控（待审核 → 正常 / 封禁）、风控字段（风险等级、黑名单、投诉次数）。

1.2 注册 ≠ 认证，但本次合并处理

理想架构里"账号注册"和"企业认证"是两件事。但在 MVP 阶段，我们把企业基础信息 + 资质材料 + 账号初始化合并到一次提交里，由后端在一个数据库事务中落三张表，注册成功即进入 PENDING_AUDIT（待审核） 状态，审核通过后才转为 NORMAL（正常）。

1.3 核心流程

 企业填写表单 + 上传三证
        │
        ▼
 POST /enterprise/save (multipart/form-data)
        │
        ├─ 1. 创建企业主表 t_enterprise（状态=待审核, code=uuid）
        ├─ 2. 创建企业信息表 t_enterprise_info（基础工商信息）
        ├─ 3. 三张图片上传阿里云 OSS → 拿到访问 URL
        └─ 4. 创建资质材料表 t_enterprise_qualification（URL + 联系人 + 机构代码）
        │
        ▼
 返回 {code:1, message:"保存成功"}，前端进入"审核中"页

二、数据库设计

2.1 整体思路：一主两从，按"稳定性"分表

我把企业数据按变更频率和用途拆成三张表，而不是一股脑塞进一张大表：

拆表的好处：主表负责"账号生命周期与风控"，信息表负责"工商档案"，资质表负责"审核材料"。审核流、风控流只碰主表，不会和海量工商字段耦合。

设计取舍：三张表之间我用 enterprise_id（普通 Int 字段）关联，而非数据库外键。理由是审核业务里"先建主表拿 ID、再建子表"的时序清晰，且避免外键带来的级联锁和问题。代价是失去数据库层的引用完整性，需在应用层保证 enterprise_id 一定存在（靠事务兜底）。

2.2 企业主表 t_enterprise

 class Enterprise(Model):
     id = fields.IntField(pk=True)
     enterprise_name = fields.CharField(max_length=255)
     enterprise_code = fields.CharField(max_length=100, unique=True)  # uuid 生成
     city = fields.ForeignKeyField("models.City", null=True, on_delete=fields.RESTRICT)
     account_status = fields.IntEnumField(enum_type=AccountStatus)     # 0正常/1待审核/2封禁
     create_time = fields.DatetimeField(auto_now_add=True)
     auth_time = fields.DatetimeField(null=True)
     auth_type = fields.IntEnumField(null=True, enum_type=AuthType)
     risk_level = fields.IntEnumField(null=True, enum_type=RiskLevel)
     blacklist_status = fields.IntEnumField(enum_type=BlackListStatus)
     complaint_count = fields.IntField(default=0)
     company_website = fields.CharField(max_length=512, null=True)
     email = fields.CharField(max_length=128, null=True)
     audit_type = fields.IntEnumField(enum_type=AuditType)
     submit_time = fields.DatetimeField(null=True)

2.3 企业信息表 t_enterprise_info

 class EnterpriseInfo(Model):
     id = fields.IntField(pk=True)
     enterprise_name = fields.CharField(max_length=255)
     unified_social_credit_code = fields.CharField(max_length=50, unique=True)  # 统一社会信用代码唯一
     legal_representative = fields.CharField(max_length=100)
     registered_capital = fields.CharField(max_length=100)
     establish_date = fields.DateField(null=True)
     register_status = fields.IntEnumField(enum_type=RegisterStatus)   # 存续/在业/开业…
     industry = fields.ForeignKeyField("models.IndustryPosition", on_delete=fields.RESTRICT)
     company_scale = fields.IntEnumField(enum_type=CompanyScale)
     financing_stage = fields.IntEnumField(enum_type=FinancingStage, null=True)
     headquarters_address = fields.CharField(max_length=512)
     business_scope = fields.TextField()
     enterprise_id = fields.IntField(null=True)

2.4 资质材料表 t_enterprise_qualification

 class EnterpriseQualification(Model):
     id = fields.IntField(pk=True)
     contact_name = fields.CharField(max_length=100)
     contact_phone = fields.CharField(max_length=32)
     contact_email = fields.CharField(max_length=128)
     business_license_url = fields.CharField(max_length=512, null=True)
     legal_id_front_url = fields.CharField(max_length=512, null=True)
     legal_id_back_url = fields.CharField(max_length=512, null=True)
     org_code_cert = fields.CharField(max_length=256, null=True)
     enterprise_id = fields.IntField(null=True)

2.5 枚举统一用 IntEnum

所有"状态类"字段都用 IntEnum 而非字符串，存库是 int、代码里是语义化枚举，兼顾性能和可读性：

 class AccountStatus(IntEnum):
     NORMAL = 0          # 正常
     PENDING_AUDIT = 1   # 待审核
     BANNED = 2          # 封禁
 ​
 class RegisterStatus(IntEnum):
     CONTINUE = 1        # 存续
     IN_BUSINESS = 2     # 在业
     OPEN = 3            # 开业
     SUSPENDED = 4       # 停业
     LIQUIDATION = 5     # 清算
     REVOKED = 6         # 吊销
     CANCELLED = 7       # 注销
 ​
 class CompanyScale(IntEnum):
     SCALE_0_20 = 1
     SCALE_20_99 = 2
     SCALE_100_499 = 3
     SCALE_500_999 = 4
     SCALE_1000_9999 = 5
     SCALE_10000_PLUS = 6

前端下拉框直接绑定枚举 int 值即可，避免了"中文文案入库"带来的脏数据和排序难题。

三、开发流程（分层架构）

模块采用清晰的四层：Router（路由） → Dependency（表单解析） → Schema（校验） → Service（业务） → Model（落库）。

3.1 接口层：纯路由，不含逻辑

 @enterprise_router.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="法人身份证（反面）"),
 ):
     from app.services.enterrise_service import EnterpriseService
     await EnterpriseService.saveEnterpriseInfo(
         form, business_license_file, legal_id_front_file, legal_id_back_file
    )
     return {"code": 1, "message": "保存成功"}

要点：

• 用 multipart/form-data 同时收文本字段 + 文件，前端用 FormData 提交；

• 文本字段交给 Depends(get_enterprise_form) 解析，文件用 File(…) 接收；

• 返回统一结构 {code, message}，与前端约定一致。

3.2 依赖层：把 Form 参数映射成强类型 Schema

FastAPI 的 Form 依赖把散落的表单字段聚合成一个 Pydantic 模型，路由里直接拿到类型安全的 form 对象：

 
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="成立日期"),
     register_status=Form(…, description="登记状态"),
     industry_id=Form(…, description="所属行业"),
     company_scale=Form(…, description="公司规模"),
     financing_stage=Form(None, description="融资阶段(可为空)"),
     headquarters_address=Form(…, description="总部地址"),
     business_scope=Form(…, description="经营范围"),
     contact_name=Form(…, description="联系人姓名"),
     contact_phone=Form(…, description="联系电话"),
     contact_email=Form(…, description="联系邮箱"),
     org_code_cert=Form(None, description="组织机构代码证"),
 ):
     return EnterpriseCreateRequest(
         enterprise_name=enterprise_name,
         unified_social_credit_code=unified_social_credit_code,
         # … 其余字段同理
    )

3.3 服务层：事务 + OSS，保证"要么全成、要么全败"

这是模块的核心。一次提交要写三张表 + 传三张图，必须原子化：

 
@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. 信息表（拿到主表 id 回填）
         await EnterpriseInfo.create(
             enterprise_name=form.enterprise_name,
             unified_social_credit_code=form.unified_social_credit_code,
             # … 其余字段
             enterprise_id=enterprise.id
        )
         # 3. 三张图上传 OSS
         oss = AliyunOSSTool()
         is_success, success_res = oss.upload_single_file(
             await business_license_file.read(),
             business_license_file.filename, oss_path="enterprise/")
         if not is_success:
             raise Exception("上传营业执照图片失败")
         # … 身份证正反面同理
         # 4. 资质表（写入 OSS 返回的访问 URL）
         await EnterpriseQualification.create(
             contact_name=form.contact_name,
             contact_phone=form.contact_phone,
             contact_email=form.contact_email,
             org_code_cert=form.org_code_cert,
             enterprise_id=enterprise.id,
             business_license_url=success_res["access_url"],
             legal_id_front_url=success_res2["access_url"],
             legal_id_back_url=success_res3["access_url"],
        )

关键点：整段包在 async with in_transaction() 里。只要任何一步（尤其是 OSS 上传）抛异常，事务整体回滚，不会出现"主表建了、图片没传、资质表缺 URL"的脏数据。

四、设计亮点与踩坑记录

4.1 亮点：状态机从注册起就介入

account_status 在创建主表时直接写入 PENDING_AUDIT，而不是先建个"正常"账号再改。这意味着企业从诞生起就处于平台管控之下，审核通过才放行，天然防住"未审先发"。

4.2 亮点：企业 Code 用 UUID，不暴露自增 ID

对外暴露的 enterprise_code 是 uuid4()，避免用自增主键做业务标识带来的枚举遍历风险。

4.3 踩坑：Pydantic v2 的 str = Field(None) 是个陷阱

我最初把可选文本字段写成：

org_code_cert: str = Field(None, description="组织机构代码证") # ❌

在 Pydantic v2 里，类型是 str（必填），却给了默认 None。当前端没传这个 key 时，Pydantic 用默认值 None 去校验 str 类型，直接报：

org_code_cert
Input should be a valid string [type=string_type, input_value=None]

正确写法是显式声明可空：

from typing import Optional
org_code_cert: Optional[str] = Field(None, description="组织机构代码证") # ✅

同理，establish_date: Optional[date]、financing_stage: Optional[int] 如果允许留空，也都必须加 Optional，否则前端省略该字段就会 422。结论：只要字段可能不传，类型就必须是 Optional[X]。

4.4 踩坑：前端"条件 append"会让可选字段变 None

前端用 if (value) formData.append(key, value) 时，空字符串 '' 是 falsy，字段会被整个省略。后端收不到 key → 命中上面的 None 校验。改成始终 append 空串 formData.append('org_code_cert', value || '') 即可规避。

4.5 注意：文件上传建议限定类型与大小

当前 business_license_file 等是 File(None)，未做后缀/大小校验。生产环境应在依赖层或中间件加 content_type 白名单和 Content-Length 上限，防止超大文件或恶意脚本拖垮 OSS。

五、总结

企业端注册模块看似只是"一个表单提交"，背后要兼顾资质留存、风控状态、多表一致性、文件存储四件事。本文的核心经验：

下一篇可以展开写"审核流"——管理员如何在管理端查看待审企业、通过/驳回并驱动 account_status 与 auth_time 变更。欢迎在评论区交流。