管理者端-查询企业列表

apis层

@enterprise_router.get("/list", summary="查询企业列表", description="查询企业列表") async def select_enterprise_list(page: int = Query(1, title="页码", description="页码", ge=1), page_size: int = Query(10, title="每页数量", description="每页数量", ge=10, le=50), enterprise_name: str = Query(None, title="企业名称", description="企业名称"), submit_time_start: str = Query(None, title="提交时间开始", description="提交时间开始"), submit_time_end: str = Query(None, title="提交时间结束", description="提交时间结束") ): res=await 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最小 1，page_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): query=Enterprise.all() if enterprise_name: query=query.filter(enterprise_name__contains=enterprise_name) if submit_time_start: query=query.filter(submit_time__gte=submit_time_start) if submit_time_end: query=query.filter(submit_time__lte=submit_time_end) total_count=await query.count() total_page=math.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_id=enterprise_id).prefetch_related("industry") enterprise_qualification=await EnterpriseQualification.get_or_none(enterprise_id=enterprise_id) enterprise_review=await EnterpriseReview.get_or_none(enterprise_id=enterprise_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): res=await EnterpriseService.select_enterprise_by_id(enterprise_id) return { "code": 1, "message": "查询成功", "data": res } @staticmethod async def select_enterprise_by_id(enterprise_id: int): enterprise=await Enterprise.get_or_none(id=enterprise_id) if not enterprise: raise Exception("企业不存在") enterpriseinfo = await EnterpriseInfo.get(enterprise_id=enterprise_id).prefetch_related("industry") enterprise_qualification=await EnterpriseQualification.get_or_none(enterprise_id=enterprise_id) enterprise_review=await EnterpriseReview.get_or_none(enterprise_id=enterprise_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_id=enterpriseReviewCreateRequest.enterprise_id enterprise_review=await EnterpriseReview.get_or_none(enterprise_id=enterprise_id) if enterprise_review is None: await EnterpriseReview.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 if enterpriseReviewCreateRequest.review_result else enterprise_review.review_result enterprise_review.review_reason=enterpriseReviewCreateRequest.review_reason if enterpriseReviewCreateRequest.review_reason else enterprise_review.review_reason enterprise_review.remark=enterpriseReviewCreateRequest.remark if enterpriseReviewCreateRequest.remark else enterprise_review.remark enterprise_review.review_time=now() await enterprise_review.save() if enterpriseReviewCreateRequest.review_result==1: enterprise=await Enterprise.get_or_none(id=enterprise_id) enterprise.account_status=AccountStatus.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): res=await EnterpriseService.login(loginMobileRequest) return { "code": 1, "message": "登录成功", "data": res } @staticmethod async def login(loginMobileRequest: LoginMobileRequest): enterprise_qualifications = await EnterpriseQualification.filter(contact_phone=loginMobileRequest.mobile) if enterprise_qualifications is None: raise Exception("手机号不存在") for enterprise_qualification in enterprise_qualifications: enterprise_id=enterprise_qualification.enterprise_id enterprise_review=await EnterpriseReview.get_or_none(enterprise_id=enterprise_id) if enterprise_review.review_result == 1: key = f"boss-api:enterprise-login:sms:{loginMobileRequest.mobile}" redis_code=redis_client.get(key) if redis_code is None: raise Exception("验证码已过期") if redis_code != loginMobileRequest.code: raise Exception("验证码错误") 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, } raise Exception("登录失败")

先用手机号去EnterpriseQualification表里找对应的企业资质记录。手机号没匹配到时，for循环一次都不执行，最后落到raise Exception("登录失败")。

接着for循环：一个手机号理论上可能关联多个企业（代码留了这个口子），所以逐个看。只有当这家企业审核通过了（review_result == 1）才让它登录。登录时去 Redis 按手机号取之前发的短信验证码：取不到说明过期了，对不上说明输错了。都过了才发access_token/refresh_token，发完顺手把 Redis 里的验证码删掉（一次性凭证，不能重复使用）。最后要是循环走完没一家满足，就抛"登录失败"。