管理者端-查詢企業(yè)列表apis層enterprise_router.get(/list, summary查詢企業(yè)列表, description查詢企業(yè)列表) async def select_enterprise_list(page: int Query(1, title頁碼, description頁碼, ge1), page_size: int Query(10, title每頁數(shù)量, description每頁數(shù)量, ge10, le50), enterprise_name: str Query(None, title企業(yè)名稱, description企業(yè)名稱), submit_time_start: str Query(None, title提交時間開始, description提交時間開始), submit_time_end: str Query(None, title提交時間結(jié)束, description提交時間結(jié)束) ): resawait EnterpriseService.select_enterprise_list(page,page_size,enterprise_name,submit_time_start,submit_time_end) return { code: 1, message: 查詢成功, data: res }GET /list就是一個帶篩選的分頁查詢。分頁參數(shù)page/page_size走Query還加了邊界page最小 1page_size卡在 10~50 之間——這等于硬性規(guī)定每頁至少 10 條、最多 50 條前端想一次拉 100 條是拉不到的。enterprise_name和起止時間都是可選篩選條件None表示不篩。路由本身不碰數(shù)據(jù)庫把參數(shù)原樣丟給 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 }先把查詢條件動態(tài)拼上去有企業(yè)名就按contains模糊匹配填了起止時間就按submit_time區(qū)間過濾。剩下的交給count算出總數(shù)再用math.ceil算總頁數(shù)total_count / page_size向上取整避免出現(xiàn)半頁算不出的情況。分頁用的是offset((page-1)*page_size).limit(page_size)——也就是跳過前面幾頁只取當(dāng)前頁那幾條。真正的重頭戲在for循環(huán)里每家企業(yè)要把它的工商信息、資質(zhì)、審核記錄三張表都撈出來拼一起行業(yè)名稱從industry關(guān)聯(lián)里取取不到就填None。最后返回的不只是列表還把總數(shù)、總頁數(shù)、當(dāng)前頁碼、每頁大小都帶回去前端做分頁條直接用。管理者端-企業(yè)詳情enterprise_router.get(/detail/{enterprise_id}, summary查詢企業(yè)詳情, description查詢企業(yè)詳情) 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(企業(yè)不存在) 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 找企業(yè)主體找不到直接報錯。然后一次性把工商信息順帶把industry關(guān)聯(lián)預(yù)加載出來省得后面再查一次、資質(zhì)、審核記錄都查出來拼成同一個結(jié)構(gòu)返回。跟列表不一樣的地方這里工商信息用的是EnterpriseInfo.get(...)不是get_or_none意味著它默認(rèn)這家企業(yè)一定有對應(yīng)的工商信息末尾enterpriseinfo.industry.name也是直接取沒做空判斷。所以這段代碼隱含一個前提——能進審核流程的企業(yè)industry一定存在。管理者端-企業(yè)狀態(tài)審核enterprise_router.post(/review, summary企業(yè)審核, description企業(yè)審核) 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查有沒有沒有就新建填上結(jié)果、理由、備注和審核時間已經(jīng)有了就走更新分支——注意那個新值 if 新值 else 舊值的寫法意思是前端沒傳空值的字段就不覆蓋庫里原來的值只改傳了的。不管新建還是更新審核時間都刷新成當(dāng)下。最后if review_result 1是關(guān)鍵聯(lián)動審核通過1才把企業(yè)的account_status置為NORMAL正常企業(yè)這下才能真正登錄使用。審核不通過review_result不是 1就不動賬號狀態(tài)企業(yè)繼續(xù)卡在待審核。企業(yè)端-登錄enterprise_router.post(/login, summary企業(yè)登錄, description企業(yè)登錄) 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表里找對應(yīng)的企業(yè)資質(zhì)記錄。手機號沒匹配到時for循環(huán)一次都不執(zhí)行最后落到raise Exception(登錄失敗)。接著for循環(huán)一個手機號理論上可能關(guān)聯(lián)多個企業(yè)代碼留了這個口子所以逐個看。只有當(dāng)這家企業(yè)審核通過了review_result 1才讓它登錄。登錄時去 Redis 按手機號取之前發(fā)的短信驗證碼取不到說明過期了對不上說明輸錯了。都過了才發(fā)access_token/refresh_token發(fā)完順手把 Redis 里的驗證碼刪掉一次性憑證不能重復(fù)使用。最后要是循環(huán)走完沒一家滿足就拋登錄失敗。