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