實(shí)現(xiàn)了企業(yè)列表分頁查詢、企業(yè)詳情獲取、企業(yè)審核狀態(tài)變更及企業(yè)驗(yàn)證碼登錄四個(gè)核心接口二、接口層實(shí)現(xiàn)Router接口層負(fù)責(zé)定義路由、接收請(qǐng)求參數(shù)、調(diào)用服務(wù)層并返回統(tǒng)一響應(yīng)格式。1. 獲取所有企業(yè)列表python運(yùn)行enterprise_router.get(path/getenterprises/, summary獲取所有企業(yè)) async def getenterprises( page: int 1, size: int 3, name: str | None None, submit_start: str | None None, submit_end: str | None None ): res await EnterPriseService.getenterprises(page, size, name, submit_start, submit_end) return {code: 200, data: res, msg: 獲取成功}功能支持分頁查詢、企業(yè)名稱模糊搜索、提交時(shí)間范圍篩選參數(shù)page頁碼、size每頁條數(shù)、name企業(yè)名、submit_start/end時(shí)間范圍響應(yīng)統(tǒng)一返回code/data/msg格式2. 獲取單個(gè)企業(yè)信息python運(yùn)行enterprise_router.get(path/getenterpriseinfo/, summary獲取單個(gè)企業(yè)信息) async def getenterpriseinfo(id: int): res await EnterPriseService.getenterpriseinfo(id) return {code: 200, data: res, msg: 獲取成功}功能根據(jù)企業(yè) ID 獲取完整企業(yè)信息基礎(chǔ)信息 詳情 資質(zhì) 審核記錄參數(shù)id企業(yè) ID校驗(yàn)服務(wù)層會(huì)校驗(yàn)企業(yè)及關(guān)聯(lián)數(shù)據(jù)是否存在3. 企業(yè)審核python運(yùn)行enterprise_router.post(path/enterprisecheck/, summary企業(yè)審核) async def enterprisecheck(id: int, request: EnterpirseCheck): res await EnterPriseService.enterprisecheck(id, request) return {code: 200, data: res, msg: 操作成功}功能管理員審核企業(yè)資質(zhì)更新企業(yè)狀態(tài)并記錄審核日志參數(shù)id企業(yè) ID、request審核請(qǐng)求體包含審核結(jié)果、原因等事務(wù)服務(wù)層使用事務(wù)保證數(shù)據(jù)一致性4. 企業(yè)驗(yàn)證碼登錄python運(yùn)行enterprise_router.post(path/enterpriseloginsms/, summary企業(yè)驗(yàn)證碼登錄) async def enterpriseloginsms(request: Request): temp await request.json() res await EnterPriseService.enterpriseloginsms(temp) return {code: 200, data: res, msg: 登錄成功}功能企業(yè)用戶通過手機(jī)號(hào) 驗(yàn)證碼登錄生成 JWT Token參數(shù)request包含phone和code的 JSON 請(qǐng)求體校驗(yàn)校驗(yàn)手機(jī)號(hào)存在性、驗(yàn)證碼有效性、企業(yè)審核狀態(tài)三、服務(wù)層實(shí)現(xiàn)Service服務(wù)層封裝核心業(yè)務(wù)邏輯與數(shù)據(jù)庫交互處理復(fù)雜查詢和事務(wù)。1. 企業(yè)列表查詢getenterprisespython運(yùn)行staticmethod async def getenterprises(page: int, size: int, name: str, submit_start: str, submit_end: str): query Q() if name: query Q(enterprise_name__icontainsname) # 模糊搜索 if submit_start and submit_end: query Q(submit_time__range(submit_start, submit_end)) # 時(shí)間范圍篩選 result [] # 分頁查詢企業(yè)基礎(chǔ)信息 enterprise await Enterprise.filter(query).offset((page - 1)* size).limit(size) for i in enterprise: temp {} temp[enterprise] dict(i) # 關(guān)聯(lián)查詢企業(yè)詳情和資質(zhì)信息 enterprise_info await EnterpriseInfo.get_or_none(enterprise_idi.id) enterprise_qualification await EnterpriseQualification.get_or_none(enterprise_idi.id) temp[enterprise_info] dict(enterprise_info) temp[enterprise_qualification] dict(enterprise_qualification) result.append(temp) # 返回分頁數(shù)據(jù)和總數(shù) return {result: result, total: await Enterprise.filter(query).count()}核心邏輯使用Q()構(gòu)建動(dòng)態(tài)查詢條件分頁offset((page-1)*size).limit(size)多表關(guān)聯(lián)遍歷企業(yè)數(shù)據(jù)關(guān)聯(lián)查詢EnterpriseInfo和EnterpriseQualification統(tǒng)計(jì)總數(shù)用于前端分頁組件2. 企業(yè)詳情查詢getenterpriseinfopython運(yùn)行staticmethod async def getenterpriseinfo(id: int): enterprise await Enterprise.get_or_none(idid) if not enterprise: raise Exception(企業(yè)不存在) # 預(yù)加載行業(yè)信息避免N1查詢 enterprise_info await EnterpriseInfo.get_or_none(enterprise_idid).prefetch_related(industry) if not enterprise_info: raise Exception(企業(yè)信息不存在) enterprise_qualification await EnterpriseQualification.get_or_none(enterprise_idid) if not enterprise_qualification: raise Exception(企業(yè)資質(zhì)不存在) enterprise_record await EnterpriseRecord.filter(enterprise_idid) result { enterprise: dict(enterprise), enterprise_info: dict(enterprise_info), enterprise_qualification: dict(enterprise_qualification), enterprise_record: [dict(i) for i in enterprise_record], industryname: enterprise_info.industry.name # 直接獲取預(yù)加載的行業(yè)名稱 } return result核心優(yōu)化使用prefetch_related(industry)預(yù)加載關(guān)聯(lián)的行業(yè)數(shù)據(jù)避免循環(huán)查詢多層校驗(yàn)確保企業(yè)及關(guān)聯(lián)數(shù)據(jù)完整存在結(jié)構(gòu)化返回將多表數(shù)據(jù)整合成統(tǒng)一字典方便前端使用3. 企業(yè)審核enterprisecheckpython運(yùn)行staticmethod async def enterprisecheck(id: int, request: EnterpirseCheck): enterpriseobj await Enterprise.get_or_none(idid) if not enterpriseobj: raise Exception(企業(yè)不存在) recordobj await EnterpriseReview.get_or_none(enterprise_idid) # 開啟事務(wù)保證數(shù)據(jù)一致性 async with in_transaction() as conn: if request.review_result 1: # 審核通過更新企業(yè)狀態(tài)為啟用 enterpriseobj.account_status 0 await enterpriseobj.save(using_dbconn) # 處理審核記錄新增或更新 if not recordobj: await EnterpriseReview.create( enterprise_idid, review_resultrequest.review_result, review_reasonrequest.review_reason, remarkrequest.remark, review_timetime.strftime(%Y-%m-%d %H:%M:%S, time.localtime()), using_dbconn ) else: request_data request.model_dump(exclude_unsetTrue) request_data[review_time] time.strftime(%Y-%m-%d %H:%M:%S, time.localtime()) await recordobj.update_from_dict(request_data) await recordobj.save(using_dbconn) # 記錄審核操作日志 await EnterpriseRecord.create( enterprise_idid, content審核通過 if request.review_result 1 else 審核未通過, record_timetime.strftime(%Y-%m-%d %H:%M:%S, time.localtime()) ) return True核心特性事務(wù)管理in_transaction()確保審核狀態(tài)變更、審核記錄、操作日志三步操作要么全部成功要么全部回滾狀態(tài)變更審核通過時(shí)將企業(yè)account_status設(shè)為 0啟用日志記錄通過EnterpriseRecord記錄審核操作便于追溯4. 企業(yè)驗(yàn)證碼登錄enterpriseloginsmspython運(yùn)行staticmethod async def enterpriseloginsms(request: Request): phone request.get(phone) enterprise_info await EnterpriseQualification.get_or_none(contact_phonephone) if not enterprise_info: raise Exception(手機(jī)號(hào)不存在) # 從Redis獲取驗(yàn)證碼 key fp4boss:sms_code:enterprise_login:{phone} code_redis r.get(key) if not code_redis: raise Exception(驗(yàn)證碼已過期) if code_redis ! request.get(code): raise Exception(驗(yàn)證碼錯(cuò)誤) # 校驗(yàn)企業(yè)審核狀態(tài) enterprisereviewobj await EnterpriseReview.get_or_none(enterprise_identerprise_info.enterprise_id) if not enterprisereviewobj: raise Exception(企業(yè)還未審核) if enterprisereviewobj.review_result ! 1: raise Exception(企業(yè)審核未通過) # 生成JWT Token enterpriseobj await EnterpriseInfo.get_or_none(enterprise_identerprise_info.enterprise_id) access_token, refresh_token create_tokens(str(enterpriseobj.enterprise_id), enterpriseobj.enterprise_name) return { enterprise_access_token: access_token, enterprise_refresh_token: refresh_token }核心流程校驗(yàn)手機(jī)號(hào)是否存在于企業(yè)資質(zhì)表從 Redis 校驗(yàn)驗(yàn)證碼有效性和時(shí)效性校驗(yàn)企業(yè)是否已通過審核生成access_token和refresh_token用于后續(xù)接口鑒權(quán)安全設(shè)計(jì)驗(yàn)證碼存儲(chǔ)在 Redis 并設(shè)置過期時(shí)間避免暴力破解四、關(guān)鍵技術(shù)點(diǎn)總結(jié)表格技術(shù)點(diǎn)說明動(dòng)態(tài)查詢使用Q()對(duì)象構(gòu)建多條件組合查詢支持模糊搜索和時(shí)間范圍篩選分頁實(shí)現(xiàn)Tortoise-ORM 的offset()和limit()方法實(shí)現(xiàn)高效分頁預(yù)加載優(yōu)化prefetch_related()解決 N1 查詢問題提升多表查詢性能事務(wù)管理in_transaction()保證復(fù)雜業(yè)務(wù)操作的數(shù)據(jù)一致性JWT 認(rèn)證登錄成功后生成雙 Token實(shí)現(xiàn)無狀態(tài)鑒權(quán)統(tǒng)一響應(yīng)所有接口返回{code: 200, data: ..., msg: ...}格式便于前端處理五、學(xué)習(xí)收獲架構(gòu)分層進(jìn)一步理解了 FastAPI 中接口層與服務(wù)層分離的優(yōu)勢業(yè)務(wù)邏輯復(fù)用性更強(qiáng)ORM 高級(jí)用法掌握了動(dòng)態(tài)查詢、分頁、預(yù)加載、事務(wù)等 Tortoise-ORM 核心特性安全設(shè)計(jì)學(xué)習(xí)了驗(yàn)證碼登錄、JWT 鑒權(quán)等安全機(jī)制的實(shí)現(xiàn)數(shù)據(jù)完整性通過多層校驗(yàn)和事務(wù)保證企業(yè)數(shù)據(jù)的完整性和一致性