FastAPI + Tortoise-ORM 企业端实战:企业认证、信息保存、审核与登录的设计与实现
- 一、前言介绍
- 1.1 项目背景
- 1.2 功能概览
- 1.3 数据模型总览
- 二、环境准备
- 2.1 依赖库清单
- 2.2 数据库与缓存配置要点
- 2.3 路由与目录结构
- 三、知识点讲解
- 3.1 企业域四表拆分与 enterprise_id 关联
- 3.2 枚举驱动的状态机
- 3.3 Form 依赖注入:表单 + 文件混合提交
- 3.4 多文件 OSS 上传与事务边界
- 3.5 列表分页与关联预加载
- 四、代码逻辑拆解
- 4.1 保存企业信息(事务 + 三文件 + 三表)
- 4.2 企业列表(分页 + 筛选 + 关联)
- 4.3 企业详情
- 4.4 企业审核(通过改状态)
- 4.5 企业登录(手机号验证码 + 审核态 + JWT)
一、前言介绍
1.1 项目背景
本项目是一个仿 BOSS 直聘的招聘平台,除求职者端之外,还包含企业(招聘方)端。企业端的核心闭环是:企业提交认证材料 → 平台审核 → 审核通过后企业可用手机号登录。本文聚焦企业域的四张表与五个接口,逐行拆解其设计与实现。
1.2 功能概览
- 保存企业信息:一次性提交企业工商信息 + 三份资质文件(营业执照、法人身份证正反面),落三张表。
- 企业列表:分页 + 按名称模糊、按提交时间区间筛选,列表项聚合主表 / 信息表 / 资质表 / 行业名称。
- 企业详情:按 ID 查出完整企业档案。
- 企业审核:对某企业标记通过 / 驳回,审核通过时把主表账号状态置为正常。
- 企业登录:用资质表里登记的手机号 + 短信验证码登录,前提是该企业已审核通过,成功后签发双 Token。
1.3 数据模型总览
Enterprise(企业主表:名称、Code、账号状态、认证类型、风险、黑名单、审核类型) │ enterprise_id(IntField 关联,非外键) ├── EnterpriseInfo(企业信息表:信用代码、法人、资本、行业、规模、融资、经营范围) ├── EnterpriseQualification(资质材料表:联系人、三文件 URL、组织机构代码证) └── EnterpriseReview(企业审核表:结果、原因、备注、审核人、时间)关键点:四张表通过enterprise_id这个整数字段手动关联,而不是用ForeignKeyField串起来(模型里外键写法被注释掉了)。这是本模块最值得讲的设计取舍,详见问题排查 5.1。
二、环境准备
2.1 依赖库清单
tortoise-orm+aerich:提供异步 ORM 功能与数据库迁移支持。pydantic:用于请求参数校验与表单(Form)数据验证。python-multipart:由于“保存企业信息”接口需要同时接收表单字段与文件,此库为必需依赖。oss2:用于将三份企业资质文件上传并存储至阿里云 OSS。redis:用于存储企业登录所需的短信验证码,并指定使用 Redis 的db=6进行隔离。
2.2 数据库与缓存配置要点
redis_client=redis.Redis(host="localhost",port=6379,db=6,decode_responses=True)- 企业登录验证码单独落在 Redis
db=6,与简历域字典树缓存(db=2)隔离,避免键空间互相污染; - 发送验证码的一端也必须写
db=6,否则读写错位,验证码永远取不到(见问题排查 5.6)。
2.3 路由与目录结构
企业端同样遵循分层约定:
app/ ├── models/enterprise.py # 四张表 + 全部枚举定义 ├── schemas/enterprise.py # 保存/审核/登录请求体 ├── apis/enterprise_api.py # /enterprise 路由组 ├── services/enterprise_service.py # 企业业务逻辑 └── core/depends.py # get_enterprise_form(表单依赖)路由统一挂在prefix="/enterprise"的APIRouter下。
三、知识点讲解
3.1 企业域四表拆分与 enterprise_id 关联
企业档案天然分成"身份状态"“工商信息”“资质材料”"审核流水"四部分,用四张表承载:
classEnterprise(Model):enterprise_code=fields.CharField(max_length=100,unique=True,description="企业Code")account_status=fields.IntEnumField(enum_type=AccountStatus,...)# ... 状态相关字段classEnterpriseInfo(Model):unified_social_credit_code=fields.CharField(max_length=50,unique=True,...)enterprise_id=fields.IntField(null=True,description="企业ID")# ... 工商信息字段classEnterpriseQualification(Model):enterprise_id=fields.IntField(null=True,description="企业ID")# ... 资质字段classEnterpriseReview(Model):enterprise_id=fields.IntField(null=True,description="企业ID")# ... 审核字段- 主表
Enterprise存"账号状态类"数据,其余三张表各存一类信息,都通过enterprise_id(整型)指回主表; - 模型里原本有
ForeignKeyField一对一的写法,被注释掉了,改用IntField手动关联——好处是建表没有外键约束、迁移简单,代价是跨表关系要业务代码自己维护(见 5.1)。
3.2 枚举驱动的状态机
账号状态、认证类型、审核类型等都用IntEnum表达,保证数据库里存的是数字、代码里读的是语义:
classAccountStatus(IntEnum):NORMAL=0# 正常PENDING_AUDIT=1# 待审核BANNED=2# 封禁classAuthType(IntEnum):NO_AUTH=0ENTERPRISE_AUTH=1LICENSE_AUTH=2ENTERPRISE_LICENSE_AUTH=3- 新建企业时直接写
account_status=AccountStatus.PENDING_AUDIT,自文档化,避免魔法数字; - 审核通过后把状态改成
AccountStatus.NORMAL,状态流转清晰可追溯。
3.3 Form 依赖注入:表单 + 文件混合提交
保存企业信息既要传一堆表单字段,又要传三个文件,请求体是multipart/form-data。这里用"Form 依赖"把字段重组回 Pydantic 模型:
asyncdefsaveEnterpriseInfo(form:EnterpriseCreateRequest=Depends(get_enterprise_form),business_license_file:UploadFile=File(None,...),legal_id_front_file:UploadFile=File(None,...),legal_id_back_file:UploadFile=File(None,...)):asyncdefget_enterprise_form(enterprise_name=Form(...,description="企业名称"),unified_social_credit_code=Form(...,description="统一社会信用代码"),# ... 其余字段逐个 Form(...)):returnEnterpriseCreateRequest(enterprise_name=enterprise_name,...)- 路由把结构化字段交给
Depends(get_enterprise_form),由它把十几个Form(...)参数组装成EnterpriseCreateRequest,文件则单独用File(...)接收; - 这种写法让"表单字段"和"文件"在路由里各司其职,避免手写解析
request.form()。
3.4 多文件 OSS 上传与事务边界
保存企业涉及三张表写入 + 三份文件上传,天然需要事务保证一致性:
asyncwithin_transaction()asconn:enterprise=awaitEnterprise.create(...)# 主表awaitEnterpriseInfo.create(...,enterprise_id=enterprise.id)# 信息表# 三份文件逐个读 + 上传 OSS# 资质表 create,带上三个 access_urlin_transaction()保证三张表要么全写、要么全滚;- 但 OSS 上传发生在数据库事务之内,它并不受数据库回滚约束(见 5.2)。
3.5 列表分页与关联预加载
列表接口要分页、要按条件筛选、还要把行业名称带出来:
query_enterprise=Enterprise.all()ifenterprise_name:query_enterprise=query_enterprise.filter(enterprise_name__icontains=enterprise_name)total_count=awaitquery_enterprise.count()offset=(page-1)*page_size query_enterprise=awaitquery_enterprise.offset(offset).limit(page_size)icontains做名称模糊匹配、gte/lte做时间区间;- 先
count()拿总数算总页数,再offset().limit()取当页,是标准游标分页写法。
四、代码逻辑拆解
4.1 保存企业信息(事务 + 三文件 + 三表)
asyncwithin_transaction()asconn: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(),auth_type=AuthType.ENTERPRISE_LICENSE_AUTH,)- 第 1 行:开启数据库事务,后续写操作都走这条连接;
- 第 3 行:
enterprise_code用uuid.uuid4()生成全局唯一码,作为企业对外标识(对外暴露 UUID 比暴露自增 ID 更安全); - 第 4–7 行:新建即进入"待审核"状态、未拉黑、审核类型为"新企业认证",并打上提交时间——状态机初始态在创建时一次性定好。
awaitEnterpriseInfo.create(enterprise_name=form.enterprise_name,unified_social_credit_code=form.unified_social_credit_code,# ... 工商字段industry_id=form.industry_id,enterprise_id=enterprise.id,)- 信息表用
enterprise_id=enterprise.id指回主表(注意传的是整型 id,不是 ORM 对象,因为这里刻意没用外键); industry_id直接存行业字典的 ID,行业名称在列表/详情时再prefetch_related取。
business_license_file_file_content=awaitbusiness_license_file.read()oss=AliyunOSSTool()is_success,success_res=oss.upload_single_file(business_license_file_file_content,business_license_file.filename,oss_path="enterprise/",)ifnotis_success:raiseException("上传营业执照图片失败")- 三份文件逻辑完全一致:先
await file.read()读成 bytes(异步必须 await),再调 OSS 工具上传到enterprise/目录,失败直接抛错; - 抛错会让
in_transaction()回滚前面两张表的写入,但已传成功的 OSS 文件回不去(见 5.2)。
awaitEnterpriseQualification.create(contact_name=form.contact_name,contact_phone=form.contact_phone,contact_email=form.contact_email,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"],)- 资质表把三份文件的
access_url与联系人信息一起入库,enterprise_id同样手动关联主表。
4.2 企业列表(分页 + 筛选 + 关联)
enterpriseinfo=awaitEnterpriseInfo.get_or_none(enterprise_id=enterprise_id).prefetch_related("industry")res_dict={"enterprise":enterprise,"enterpriseinfo":enterpriseinfo,"enterprise_qualification":enterprise_qualification,"industry":enterpriseinfo.industry.name,}get_or_none(enterprise_id=...)按整型 id 查信息表,再用.prefetch_related("industry")把行业对象一次性加载进来;- 列表项把主表、信息表、资质表、行业名称聚合成一个字典返回,前端一次拿到展示所需全部数据;
- 注意
enterpriseinfo.industry.name这行:若enterpriseinfo为None,访问.industry会直接抛AttributeError(见 5.3)。
4.3 企业详情
enterprise=awaitEnterprise.get_or_none(id=enterprise_id)enterpriseinfo=awaitEnterpriseInfo.get_or_none(enterprise_id=enterprise_id).prefetch_related("industry")enterprise_qualification=awaitEnterpriseQualification.get_or_none(enterprise_id=enterprise_id)return{"enterprise":enterprise,"enterpriseinfo":enterpriseinfo,"enterprise_qualification":enterprise_qualification,"industry":enterpriseinfo.industry.name,}- 详情是列表逻辑的精简版:按 ID 查主表 + 信息表(带行业)+ 资质表,三表聚合返回;
- 同样存在 4.2 提到的
enterpriseinfo为空时的隐患。
4.4 企业审核(通过改状态)
enterprise_review=awaitEnterpriseReview.get_or_none(enterprise_id=enterprise_id)ifenterprise_reviewisNone:awaitEnterpriseReview.create(enterprise_id=enterprise_id,review_result=enterpriseReviewCreateRequest.review_result,review_reason=enterpriseReviewCreateRequest.review_reason,remark=enterpriseReviewCreateRequest.remark,review_time=now(),)else:enterprise_review.review_result=enterpriseReviewCreateRequest.review_result \ifenterpriseReviewCreateRequest.review_resultelseenterprise_review.review_result# ... 原因、备注同样"传了才覆盖"enterprise_review.review_time=now()awaitenterprise_review.save()- 第 1–2 行:先查是否已有审核记录,没有就新建,有就更新,实现审核流水可重写;
- 第 7–8 行:更新用
if 新值 else 旧值的写法,前端没传的字段保留原值,避免被None覆盖; - 注意这里没有用
exclude_unset,而是显式判断,因为审核请求体里review_result是必填Field(...),但原因/备注可空,需要区分"传了空"和"没传"。
ifenterpriseReviewCreateRequest.review_result==1:enterprise=awaitEnterprise.get_or_none(id=enterprise_id)enterprise.account_status=AccountStatus.NORMALawaitenterprise.save()- 审核结果为 1(通过)时,把主表
account_status从"待审核"改成"正常",状态机向前推进一步; - 驳回(非 1)则只留审核记录,不改变账号状态,企业依旧处于待审核、无法登录。
4.5 企业登录(手机号验证码 + 审核态 + JWT)
enterprise_qualifications=awaitEnterpriseQualification.filter(contact_phone=loginMobileRequest.mobile)forenterprise_qualificationinenterprise_qualifications:enterprise_id=enterprise_qualification.enterprise_id enterprise_review=awaitEnterpriseReview.get_or_none(enterprise_id=enterprise_id)ifenterprise_review.review_result==1:# 审核通过key=f"boss-api:enterprise-login:sms:{loginMobileRequest.mobile}"redis_code=redis_client.get(key)ifredis_codeisNone:raiseException("验证码已过期")ifredis_code!=loginMobileRequest.code:raiseException("验证码错误")access_token,refresh_token=create_tokens(str(enterprise_id),loginMobileRequest.mobile)redis_client.delete(key)return{"enterprise_access_token":access_token,"enterprise_refresh_token":refresh_token}raiseException("账号未审核通过")- 第 1 行:用资质表登记的
contact_phone反查企业,一个手机号可能对应多条资质记录,所以用遍历; - 第 4 行:
enterprise_review.review_result == 1是登录前提——未审核通过直接跳过,循环结束抛"账号未审核通过"; - 第 5–9 行:审核通过的才校验 Redis 短信验证码(读
db=6),一致则签发双 Token 并删除验证码,防止重放; create_tokens(str(enterprise_id), mobile):Token 的user_id实际装的是企业 ID,与求职者端共用同一套 JWT 工具。