實現了企業列表分頁查詢、企業詳情獲取、企業審核狀態變更及企業驗證碼登錄四個核心接口二、接口層實現Router接口層負責定義路由、接收請求參數、調用服務層并返回統一響應格式。1. 獲取所有企業列表python運行enterprise_router.get(path/getenterprises/, summary獲取所有企業) 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: 獲取成功}功能支持分頁查詢、企業名稱模糊搜索、提交時間范圍篩選參數page頁碼、size每頁條數、name企業名、submit_start/end時間范圍響應統一返回code/data/msg格式2. 獲取單個企業信息python運行enterprise_router.get(path/getenterpriseinfo/, summary獲取單個企業信息) async def getenterpriseinfo(id: int): res await EnterPriseService.getenterpriseinfo(id) return {code: 200, data: res, msg: 獲取成功}功能根據企業 ID 獲取完整企業信息基礎信息 詳情 資質 審核記錄參數id企業 ID校驗服務層會校驗企業及關聯數據是否存在3. 企業審核python運行enterprise_router.post(path/enterprisecheck/, summary企業審核) async def enterprisecheck(id: int, request: EnterpirseCheck): res await EnterPriseService.enterprisecheck(id, request) return {code: 200, data: res, msg: 操作成功}功能管理員審核企業資質更新企業狀態并記錄審核日志參數id企業 ID、request審核請求體包含審核結果、原因等事務服務層使用事務保證數據一致性4. 企業驗證碼登錄python運行enterprise_router.post(path/enterpriseloginsms/, summary企業驗證碼登錄) async def enterpriseloginsms(request: Request): temp await request.json() res await EnterPriseService.enterpriseloginsms(temp) return {code: 200, data: res, msg: 登錄成功}功能企業用戶通過手機號 驗證碼登錄生成 JWT Token參數request包含phone和code的 JSON 請求體校驗校驗手機號存在性、驗證碼有效性、企業審核狀態三、服務層實現Service服務層封裝核心業務邏輯與數據庫交互處理復雜查詢和事務。1. 企業列表查詢getenterprisespython運行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)) # 時間范圍篩選 result [] # 分頁查詢企業基礎信息 enterprise await Enterprise.filter(query).offset((page - 1)* size).limit(size) for i in enterprise: temp {} temp[enterprise] dict(i) # 關聯查詢企業詳情和資質信息 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) # 返回分頁數據和總數 return {result: result, total: await Enterprise.filter(query).count()}核心邏輯使用Q()構建動態查詢條件分頁offset((page-1)*size).limit(size)多表關聯遍歷企業數據關聯查詢EnterpriseInfo和EnterpriseQualification統計總數用于前端分頁組件2. 企業詳情查詢getenterpriseinfopython運行staticmethod async def getenterpriseinfo(id: int): enterprise await Enterprise.get_or_none(idid) if not enterprise: raise Exception(企業不存在) # 預加載行業信息避免N1查詢 enterprise_info await EnterpriseInfo.get_or_none(enterprise_idid).prefetch_related(industry) if not enterprise_info: raise Exception(企業信息不存在) enterprise_qualification await EnterpriseQualification.get_or_none(enterprise_idid) if not enterprise_qualification: raise Exception(企業資質不存在) 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 # 直接獲取預加載的行業名稱 } return result核心優化使用prefetch_related(industry)預加載關聯的行業數據避免循環查詢多層校驗確保企業及關聯數據完整存在結構化返回將多表數據整合成統一字典方便前端使用3. 企業審核enterprisecheckpython運行staticmethod async def enterprisecheck(id: int, request: EnterpirseCheck): enterpriseobj await Enterprise.get_or_none(idid) if not enterpriseobj: raise Exception(企業不存在) recordobj await EnterpriseReview.get_or_none(enterprise_idid) # 開啟事務保證數據一致性 async with in_transaction() as conn: if request.review_result 1: # 審核通過更新企業狀態為啟用 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核心特性事務管理in_transaction()確保審核狀態變更、審核記錄、操作日志三步操作要么全部成功要么全部回滾狀態變更審核通過時將企業account_status設為 0啟用日志記錄通過EnterpriseRecord記錄審核操作便于追溯4. 企業驗證碼登錄enterpriseloginsmspython運行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(手機號不存在) # 從Redis獲取驗證碼 key fp4boss:sms_code:enterprise_login:{phone} code_redis r.get(key) if not code_redis: raise Exception(驗證碼已過期) if code_redis ! request.get(code): raise Exception(驗證碼錯誤) # 校驗企業審核狀態 enterprisereviewobj await EnterpriseReview.get_or_none(enterprise_identerprise_info.enterprise_id) if not enterprisereviewobj: raise Exception(企業還未審核) if enterprisereviewobj.review_result ! 1: raise Exception(企業審核未通過) # 生成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 }核心流程校驗手機號是否存在于企業資質表從 Redis 校驗驗證碼有效性和時效性校驗企業是否已通過審核生成access_token和refresh_token用于后續接口鑒權安全設計驗證碼存儲在 Redis 并設置過期時間避免暴力破解四、關鍵技術點總結表格技術點說明動態查詢使用Q()對象構建多條件組合查詢支持模糊搜索和時間范圍篩選分頁實現Tortoise-ORM 的offset()和limit()方法實現高效分頁預加載優化prefetch_related()解決 N1 查詢問題提升多表查詢性能事務管理in_transaction()保證復雜業務操作的數據一致性JWT 認證登錄成功后生成雙 Token實現無狀態鑒權統一響應所有接口返回{code: 200, data: ..., msg: ...}格式便于前端處理五、學習收獲架構分層進一步理解了 FastAPI 中接口層與服務層分離的優勢業務邏輯復用性更強ORM 高級用法掌握了動態查詢、分頁、預加載、事務等 Tortoise-ORM 核心特性安全設計學習了驗證碼登錄、JWT 鑒權等安全機制的實現數據完整性通過多層校驗和事務保證企業數據的完整性和一致性