实现了企业列表分页查询、企业详情获取、企业审核状态变更及企业验证码登录四个核心接口二、接口层实现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 鉴权等安全机制的实现数据完整性通过多层校验和事务保证企业数据的完整性和一致性