管理者端-查詢企業列表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 里的驗證碼刪掉一次性憑證不能重復使用。最后要是循環走完沒一家滿足就拋登錄失敗。