一、今日完成开发功能企业列表分页查询接口企业详情查询接口企业信息审核接口企业短信验证码登录接口二、业务完整代码1. EnterpriseService 业务服务代码python运行import math import logging from app.models.enterprise import Enterprise, EnterpriseInfo, EnterpriseQualification, EnterpriseReview from app.core.constant import AccountStatus from app.core.utils import now from app.core.redis import redis_client from app.schemas.enterprise_schema import EnterpriseReviewCreateRequest, LoginMobileRequest logger logging.getLogger(__name__) class EnterpriseService: staticmethod async def select_enterprise_list(page, page_size, enterprise_name, submit_time_start, submit_time_end): query_enterprise Enterprise.all() if enterprise_name: query_enterprise query_enterprise.filter(enterprise_name__icontainsenterprise_name) if submit_time_start: query_enterprise query_enterprise.filter(submit_time__gtesubmit_time_start) if submit_time_end: query_enterprise query_enterprise.filter(submit_time__ltesubmit_time_end) total_count await query_enterprise.count() total_page math.ceil(total_count / page_size) offset (page - 1) * page_size query_enterprise query_enterprise.offset(offset).limit(page_size) enterprise_list [] for enterprise in await query_enterprise: enterprise_id enterprise.id enterpriseinfo await EnterpriseInfo.get_or_none(enterprise_identerprise_id).prefetch_related(industry) enterprise_qualification await EnterpriseQualification.get_or_none(enterprise_identerprise_id) res_dict { enterprise: enterprise, enterpriseinfo: enterpriseinfo, enterprise_qualification: enterprise_qualification, industry: enterpriseinfo.industry.name } enterprise_list.append(res_dict) return { code: 1, message: 查询成功, data: { total_count: total_count, total_page: total_page, page: page, page_size: page_size, enterprise_list: enterprise_list } } staticmethod async def select_enterprise_by_id(enterprise_id): enterprise await Enterprise.get_or_none(identerprise_id) enterpriseinfo await EnterpriseInfo.get_or_none(enterprise_identerprise_id).prefetch_related(industry) enterprise_qualification await EnterpriseQualification.get_or_none(enterprise_identerprise_id) return { code: 1, message: 查询成功, data: { enterprise: enterprise, enterpriseinfo: enterpriseinfo, enterprise_qualification: enterprise_qualification, industry: enterpriseinfo.industry.name } } staticmethod async def review_enterprise(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): enterprise_id enterpriseReviewCreateRequest.enterprise_id enterprise_review await 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_result enterpriseReviewCreateRequest.review_result if enterpriseReviewCreateRequest.review_result else enterprise_review.review_result enterprise_review.review_reason enterpriseReviewCreateRequest.review_reason if enterpriseReviewCreateRequest.review_reason else enterprise_review.review_reason enterprise_review.remark enterpriseReviewCreateRequest.remark if enterpriseReviewCreateRequest.remark else enterprise_review.remark enterprise_review.review_time now() await enterprise_review.save() if enterpriseReviewCreateRequest.review_result 1: enterprise await Enterprise.get_or_none(identerprise_id) enterprise.account_status AccountStatus.NORMAL await enterprise.save() staticmethod async def login(loginMobileRequest): enterprise_qualifications await EnterpriseQualification.filter(contact_phoneloginMobileRequest.mobile) if enterprise_qualifications is None: raise Exception(手机号不存在) for enterprise_qualification in enterprise_qualifications: enterprise_id enterprise_qualification.enterprise_id logger.info(fenterprise_id: {enterprise_id}) enterprise_review await EnterpriseReview.get_or_none(enterprise_identerprise_id) logger.info(fenterprise_review: {enterprise_review}) if enterprise_review.review_result 1: # 审核通过 key fboss-api:enterprise-login:sms:{loginMobileRequest.mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! loginMobileRequest.code: raise Exception(验证码错误) access_token, refresh_token create_token(str(enterprise_id), loginMobileRequest.mobile) logger.info(f生成 access_token:{access_token}) logger.info(f生成 refresh_token:{refresh_token}) redis_client.delete(key) return { enterprise_access_token: access_token, enterprise_refresh_token: refresh_token, } raise Exception(账号未审核通过)2. 企业路由接口 enterprise_api.pypython运行from fastapi import APIRouter, Query from app.services.enterprise_service import EnterpriseService from app.schemas.enterprise_schema import EnterpriseReviewCreateRequest, LoginMobileRequest enterprise_router APIRouter(prefix/enterprise, tags[企业管理]) enterprise_router.get(/list, summary查询企业列表, description查询企业列表) async def query_enterprise_list(page: int Query(1, title页码, description页码, ge1), page_size: int Query(10, title每页数量, description每页数量, ge10, le50), enterprise_name: str Query(None, title企业名称, description企业名称), submit_time_start: str Query(None, title提交时间开始, description提交时间开始), submit_time_end: str Query(None, title提交时间结束, description提交时间结束), ): 查询企业列表 res await EnterpriseService.select_enterprise_list(page, page_size, enterprise_name, submit_time_start, submit_time_end) return { code: 1, message: 查询成功, data: res } enterprise_router.get(/detail/{enterprise_id}, summary查询企业详情, description查询企业详情) async def query_enterprise_detail(enterprise_id: int): 查询企业详情 res await EnterpriseService.select_enterprise_by_id(enterprise_id) return { code: 1, message: 查询成功, data: res } enterprise_router.post(/review, summary企业审核, description企业审核) async def review_enterprise(enterpriseReviewCreateRequest: EnterpriseReviewCreateRequest): 企业审核 res await EnterpriseService.review_enterprise(enterpriseReviewCreateRequest) return { code: 1, message: 审核成功, data: res } enterprise_router.post(/login, summary企业登录, description企业登录) async def login(loginMobileRequest: LoginMobileRequest): res await EnterpriseService.login(loginMobileRequest) return { code: 1, message: 登录成功, data: res }3. 数据库模型 EnterpriseReviewpython运行from tortoise import Model, fields class EnterpriseReview(Model): 企业审核表 id fields.IntField(pkTrue, description主键ID) review_result fields.IntField(nullTrue, description审核结果(1:通过,2:驳回)) review_reason fields.CharField(max_length512, nullTrue, description拒绝原因) # 备注说明 remark fields.TextField(nullTrue, description备注说明) # 审核时间 audit_time fields.DatetimeField(nullTrue, description审核时间) # 审核人 audit_user fields.CharField(max_length100, nullTrue, description审核人) # 关联企业主表 Enterprise一对一关系 enterprise_id fields.IntField(nullTrue, description企业ID) class Meta: table t_enterprise_review table_description 企业审核表4. 请求实体 Schemapython运行from pydantic import BaseModel, Field class EnterpriseReviewCreateRequest(BaseModel): review_result: int Field(..., description审核结果(1:通过,2:驳回)) review_reason: str | None Field(None, description拒绝原因) # 备注说明 remark: str | None Field(None, description备注说明) # 关联企业主表 Enterprise一对一关系 enterprise_id: int Field(..., description企业ID) class LoginMobileRequest(BaseModel): mobile: str Field(..., description手机号) code: str Field(..., description验证码)