管理者端-查询企业列表apis层enterprise_router.get(/list, summary查询企业列表, description查询企业列表) async def select_enterprise_list(page: int Query(1, title页码, description页码, ge1), page_size: int Query(10, title每页数量, description每页数量, ge10, le50), enterprise_name: str Query(None, title企业名称, description企业名称), submit_time_start: str Query(None, title提交时间开始, description提交时间开始), submit_time_end: str Query(None, title提交时间结束, description提交时间结束) ): resawait EnterpriseService.select_enterprise_list(page,page_size,enterprise_name,submit_time_start,submit_time_end) return { code: 1, message: 查询成功, data: res }GET /list就是一个带筛选的分页查询。分页参数page/page_size走Query还加了边界page最小 1page_size卡在 10~50 之间——这等于硬性规定每页至少 10 条、最多 50 条前端想一次拉 100 条是拉不到的。enterprise_name和起止时间都是可选筛选条件None表示不筛。路由本身不碰数据库把参数原样丢给 Service回来包成{code, message, data}返回。services层staticmethod async def select_enterprise_list(page: int, page_size: int, enterprise_name: str, submit_time_start: str, submit_time_end: str): queryEnterprise.all() if enterprise_name: queryquery.filter(enterprise_name__containsenterprise_name) if submit_time_start: queryquery.filter(submit_time__gtesubmit_time_start) if submit_time_end: queryquery.filter(submit_time__ltesubmit_time_end) total_countawait query.count() total_pagemath.ceil(total_count/page_size) query await query.offset((page-1)*page_size).limit(page_size) enterprise_list [] for enterprise in query: enterprise_id enterprise.id enterpriseinfo await EnterpriseInfo.get_or_none(enterprise_identerprise_id).prefetch_related(industry) enterprise_qualificationawait EnterpriseQualification.get_or_none(enterprise_identerprise_id) enterprise_reviewawait EnterpriseReview.get_or_none(enterprise_identerprise_id) if enterpriseinfo and enterpriseinfo.industry: industry_name enterpriseinfo.industry.name else: industry_name None res_dict{ enterprise:enterprise, enterpriseinfo:enterpriseinfo, enterprise_qualification:enterprise_qualification, enterprise_review:enterprise_review, industry:industry_name } enterprise_list.append(res_dict) return { total_count: total_count, total_page: total_page, page: page, page_size: page_size, enterprise_list: enterprise_list }先把查询条件动态拼上去有企业名就按contains模糊匹配填了起止时间就按submit_time区间过滤。剩下的交给count算出总数再用math.ceil算总页数total_count / page_size向上取整避免出现半页算不出的情况。分页用的是offset((page-1)*page_size).limit(page_size)——也就是跳过前面几页只取当前页那几条。真正的重头戏在for循环里每家企业要把它的工商信息、资质、审核记录三张表都捞出来拼一起行业名称从industry关联里取取不到就填None。最后返回的不只是列表还把总数、总页数、当前页码、每页大小都带回去前端做分页条直接用。管理者端-企业详情enterprise_router.get(/detail/{enterprise_id}, summary查询企业详情, description查询企业详情) async def select_enterprise_by_id(enterprise_id: int): resawait EnterpriseService.select_enterprise_by_id(enterprise_id) return { code: 1, message: 查询成功, data: res } staticmethod async def select_enterprise_by_id(enterprise_id: int): enterpriseawait Enterprise.get_or_none(identerprise_id) if not enterprise: raise Exception(企业不存在) enterpriseinfo await EnterpriseInfo.get(enterprise_identerprise_id).prefetch_related(industry) enterprise_qualificationawait EnterpriseQualification.get_or_none(enterprise_identerprise_id) enterprise_reviewawait EnterpriseReview.get_or_none(enterprise_identerprise_id) return { enterprise:enterprise, enterpriseinfo:enterpriseinfo, enterprise_qualification:enterprise_qualification, enterprise_review:enterprise_review, industry:enterpriseinfo.industry.name }先按 id 找企业主体找不到直接报错。然后一次性把工商信息顺带把industry关联预加载出来省得后面再查一次、资质、审核记录都查出来拼成同一个结构返回。跟列表不一样的地方这里工商信息用的是EnterpriseInfo.get(...)不是get_or_none意味着它默认这家企业一定有对应的工商信息末尾enterpriseinfo.industry.name也是直接取没做空判断。所以这段代码隐含一个前提——能进审核流程的企业industry一定存在。管理者端-企业状态审核enterprise_router.post(/review, summary企业审核, description企业审核) async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): await EnterpriseService.enterprise_review(enterpriseReviewCreateRequest) return { code: 1, message: 审核成功 } staticmethod async def enterprise_review(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_identerpriseReviewCreateRequest.enterprise_id enterprise_reviewawait EnterpriseReview.get_or_none(enterprise_identerprise_id) if enterprise_review is None: await EnterpriseReview.create( enterprise_identerprise_id, review_resultenterpriseReviewCreateRequest.review_result, review_reasonenterpriseReviewCreateRequest.review_reason, remarkenterpriseReviewCreateRequest.remark, review_timenow() ) else: enterprise_review.review_resultenterpriseReviewCreateRequest.review_result if enterpriseReviewCreateRequest.review_result else enterprise_review.review_result enterprise_review.review_reasonenterpriseReviewCreateRequest.review_reason if enterpriseReviewCreateRequest.review_reason else enterprise_review.review_reason enterprise_review.remarkenterpriseReviewCreateRequest.remark if enterpriseReviewCreateRequest.remark else enterprise_review.remark enterprise_review.review_timenow() await enterprise_review.save() if enterpriseReviewCreateRequest.review_result1: enterpriseawait Enterprise.get_or_none(identerprise_id) enterprise.account_statusAccountStatus.NORMAL await enterprise.save()审核记录一企一条所以先按enterprise_id查有没有没有就新建填上结果、理由、备注和审核时间已经有了就走更新分支——注意那个新值 if 新值 else 旧值的写法意思是前端没传空值的字段就不覆盖库里原来的值只改传了的。不管新建还是更新审核时间都刷新成当下。最后if review_result 1是关键联动审核通过1才把企业的account_status置为NORMAL正常企业这下才能真正登录使用。审核不通过review_result不是 1就不动账号状态企业继续卡在待审核。企业端-登录enterprise_router.post(/login, summary企业登录, description企业登录) async def login(loginMobileRequest: LoginMobileRequest): resawait EnterpriseService.login(loginMobileRequest) return { code: 1, message: 登录成功, data: res } staticmethod async def login(loginMobileRequest: LoginMobileRequest): enterprise_qualifications await EnterpriseQualification.filter(contact_phoneloginMobileRequest.mobile) if enterprise_qualifications is None: raise Exception(手机号不存在) for enterprise_qualification in enterprise_qualifications: enterprise_identerprise_qualification.enterprise_id enterprise_reviewawait EnterpriseReview.get_or_none(enterprise_identerprise_id) if enterprise_review.review_result 1: key fboss-api:enterprise-login:sms:{loginMobileRequest.mobile} redis_coderedis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! loginMobileRequest.code: raise Exception(验证码错误) access_token,refresh_tokencreate_tokens(str(enterprise_id),loginMobileRequest.mobile) redis_client.delete(key) return { enterprise_access_token: access_token, enterprise_refresh_token: refresh_token, } raise Exception(登录失败)先用手机号去EnterpriseQualification表里找对应的企业资质记录。手机号没匹配到时for循环一次都不执行最后落到raise Exception(登录失败)。接着for循环一个手机号理论上可能关联多个企业代码留了这个口子所以逐个看。只有当这家企业审核通过了review_result 1才让它登录。登录时去 Redis 按手机号取之前发的短信验证码取不到说明过期了对不上说明输错了。都过了才发access_token/refresh_token发完顺手把 Redis 里的验证码删掉一次性凭证不能重复使用。最后要是循环走完没一家满足就抛登录失败。