iTi-Flask-IOT/iti/applications/routes/sys/config.py

from apiflask import APIBlueprint
from flask_jwt_extended import jwt_required
from iti.applications.extensions import db
from iti.applications.models import SysConfig, SysConfigSchema
from iti.applications.routes.sys.schemas.config import (
    SysConfigCreateRequest,
    SysConfigQuery,
    SysConfigUpdateRequest,
)
from iti.applications.common.utils import success, page, page_schema
from iti.applications.extensions import sys_log
from iti.applications.common.enums import LogType
from iti.applications.common import ModelFilter, permission
from iti.applications.common.exceptions.biz_exp import BizException
from sqlalchemy import select

bp = APIBlueprint("sys_config", __name__, url_prefix="/config", tag="系统.配置管理")


@bp.get("/list")
@jwt_required()
@bp.doc(security="JWT")
@permission("system:config:list")
@bp.input(SysConfigQuery.Schema, location="query")
@bp.output(SysConfigSchema(many=True))
def get_sys_config_list(query_data: SysConfigQuery):
    """
    获取系统配置列表
    """
    return success(get_list_or_page(query_data))


@bp.get("/page")
@jwt_required()
@bp.doc(security="JWT")
@permission("system:config:list")
@bp.input(SysConfigQuery.Schema, location="query")
@bp.output(page_schema(SysConfigSchema(many=True)))
def get_sys_config_page(query_data: SysConfigQuery):
    """
    获取系统配置分页
    """
    return page(get_list_or_page(query_data))


def get_list_or_page(query_data: SysConfigQuery):
    """
    获取系统配置列表或分页
    """
    query = select(SysConfig).order_by(SysConfig.sort.asc())
    if query_data.keyword:
        kw = ModelFilter.escape_like(query_data.keyword)
        query = query.filter(
            SysConfig.code.like(f"%{kw}%") | SysConfig.name.like(f"%{kw}%")
        )
    else:
        if query_data.name:
            query = query.filter(SysConfig.name == query_data.name)
        if query_data.code:
            query = query.filter(
                SysConfig.code.like(f"%{ModelFilter.escape_like(query_data.code)}%")
            )
        if query_data.type:
            query = query.filter(SysConfig.type == query_data.type)
        if query_data.status:
            query = query.filter(SysConfig.status == query_data.status)
        if query_data.createdAt and len(query_data.createdAt) >= 2:
            query = query.filter(
                SysConfig.created_at.between(
                    query_data.createdAt[0], query_data.createdAt[1]
                )
            )
        if query_data.updatedAt and len(query_data.updatedAt) >= 2:
            query = query.filter(
                SysConfig.updated_at.between(
                    query_data.updatedAt[0], query_data.updatedAt[1]
                )
            )
    # 基于请求参数判断分页还是列表
    if query_data.page and query_data.size:
        return db.paginate(query, page=query_data.page, per_page=query_data.size)
    else:
        return db.session.scalars(query).all()


@bp.post("")
@jwt_required()
@bp.doc(security="JWT")
@permission("system:config:create")
@bp.input(SysConfigCreateRequest, location="json")
@sys_log(
    name="创建系统配置",
    desc="创建系统配置",
    type=LogType.OPERATION,
    save_db=True,
    execute_time=True,
)
def create_sys_config(json_data: dict):
    """
    创建系统配置
    """
    sys_config = SysConfig(**json_data)
    db.session.add(sys_config)
    db.session.commit()
    return success()


@bp.put("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("system:config:edit")
@bp.input(SysConfigUpdateRequest(partial=True), location="json")
@sys_log(
    name="更新系统配置",
    desc="更新系统配置",
    type=LogType.OPERATION,
    save_db=True,
    execute_time=True,
)
def update_sys_config(id: str, json_data: dict):
    """
    更新系统配置
    """
    sys_config = db.session.scalar(select(SysConfig).filter_by(id=id))
    if not sys_config:
        raise BizException("系统配置不存在")
    for key, value in json_data.items():
        if value is not None:
            setattr(sys_config, key, value)
    db.session.commit()
    return success()


@bp.delete("/<string:id>")
@jwt_required()
@bp.doc(security="JWT")
@permission("system:config:delete")
@sys_log(
    name="删除系统配置",
    desc="删除系统配置",
    type=LogType.OPERATION,
    save_db=True,
    execute_time=True,
)
def delete_sys_config(id: str):
    """
    删除系统配置
    """
    sys_config = db.session.scalar(select(SysConfig).filter_by(id=id))
    if not sys_config:
        raise BizException("系统配置不存在")
    db.session.delete(sys_config)
    db.session.commit()
    return success()