一、今日完成开发功能
- 企业列表分页查询接口
- 企业详情查询接口
- 企业信息审核接口
- 企业短信验证码登录接口
二、业务完整代码
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__icontains=enterprise_name) if submit_time_start: query_enterprise = query_enterprise.filter(submit_time__gte=submit_time_start) if submit_time_end: query_enterprise = query_enterprise.filter(submit_time__lte=submit_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_id=enterprise_id).prefetch_related("industry") enterprise_qualification = await EnterpriseQualification.get_or_none(enterprise_id=enterprise_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(id=enterprise_id) enterpriseinfo = await EnterpriseInfo.get_or_none(enterprise_id=enterprise_id).prefetch_related("industry") enterprise_qualification = await EnterpriseQualification.get_or_none(enterprise_id=enterprise_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_id=enterprise_id) if enterprise_review is None: await EnterpriseReview.create( enterprise_id=enterprise_id, review_result=enterpriseReviewCreateRequest.review_result, review_reason=enterpriseReviewCreateRequest.review_reason, remark=enterpriseReviewCreateRequest.remark, review_time=now() ) 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(id=enterprise_id) enterprise.account_status = AccountStatus.NORMAL await enterprise.save() @staticmethod async def login(loginMobileRequest): enterprise_qualifications = await EnterpriseQualification.filter(contact_phone=loginMobileRequest.mobile) if enterprise_qualifications is None: raise Exception("手机号不存在") for enterprise_qualification in enterprise_qualifications: enterprise_id = enterprise_qualification.enterprise_id logger.info(f"enterprise_id: {enterprise_id}") enterprise_review = await EnterpriseReview.get_or_none(enterprise_id=enterprise_id) logger.info(f"enterprise_review: {enterprise_review}") if enterprise_review.review_result == 1: # 审核通过 key = f"boss-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.py
python
运行
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="页码", ge=1), page_size: int = Query(10, title="每页数量", description="每页数量", ge=10, le=50), 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. 数据库模型 EnterpriseReview
python
运行
from tortoise import Model, fields class EnterpriseReview(Model): """企业审核表""" id = fields.IntField(pk=True, description="主键ID") review_result = fields.IntField(null=True, description="审核结果(1:通过,2:驳回)") review_reason = fields.CharField(max_length=512, null=True, description="拒绝原因") # 备注说明 remark = fields.TextField(null=True, description="备注说明") # 审核时间 audit_time = fields.DatetimeField(null=True, description="审核时间") # 审核人 audit_user = fields.CharField(max_length=100, null=True, description="审核人") # 关联企业主表 Enterprise，一对一关系 enterprise_id = fields.IntField(null=True, description="企业ID") class Meta: table = "t_enterprise_review" table_description = "企业审核表"4. 请求实体 Schema
python
运行
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="验证码")
