vue-fastapi-admin/vue-fastapi-backend/module_admin/controller/metatask_controller.py

from fastapi import APIRouter, Depends, Request
from config.get_db import get_db
from module_admin.service.login_service import LoginService
from module_admin.service.metatask_service import MetataskService
from module_admin.entity.vo.metatask_vo import DeleteMetataskModel,MetataskPageObject,MetataskModel,OperaMetataskModel
from module_admin.entity.vo.dataSource_vo import ProcessDefinition,ParmScheduleVo,ProcessInstancePage
from utils.page_util import PageResponseModel
from utils.response_util import ResponseUtil
from utils.log_util import logger
from module_admin.aspect.interface_auth import CheckUserInterfaceAuth
from sqlalchemy.ext.asyncio import AsyncSession
from config.enums import BusinessType
from datetime import datetime
from module_admin.entity.vo.user_vo import CurrentUserModel
from pydantic_validation_decorator import ValidateFields
from module_admin.annotation.log_annotation import Log
from module_admin.service.config_service import ConfigService


metataskController = APIRouter(prefix='/meta/metatask', dependencies=[Depends(LoginService.get_current_user)])

@metataskController.get(
    '/list', response_model=PageResponseModel, dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:list'))]
)
async def get_system_metatask_list(
    request: Request,
    metatask_page_query: MetataskPageObject = Depends(MetataskPageObject.as_query),
    query_db: AsyncSession = Depends(get_db),
):
    # 获取分页数据
    config_page_query_result = await MetataskService.get_metatask_list_services(query_db, metatask_page_query, is_page=True)
    logger.info('获取成功')

    return ResponseUtil.success(model_content=config_page_query_result)
# 获取工作流实例
@metataskController.get(
    '/process_instances',  dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:log'))]
)
async def get_process_instances_list(
    request: Request,
    metatask_page_query: ProcessInstancePage = Depends(ProcessInstancePage),
):
    # 获取分页数据
    config_page_query_result = await MetataskService.get_process_instances_services(request, metatask_page_query)
    logger.info('获取成功')

    return ResponseUtil.success(model_content=config_page_query_result)
# 获取工作流实例节点
@metataskController.get(
    '/task_nodes/{id}',  dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:log'))]
)
async def get_task_nodes_list(
    request: Request,
    id:int,
):

    config_page_query_result = await MetataskService.get_task_nodes_services(request, id)
    logger.info('获取成功')

    return ResponseUtil.success(rows=config_page_query_result)
# 获取日志详情
@metataskController.get(
    '/log_details/{id}', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:log'))]
)
async def get_log_details(
    request: Request,
    id:int,
):
    # 获取分页数据
    config_page_query_result = await MetataskService.get_log_details_services(request,id)
    logger.info('获取成功')

    return ResponseUtil.success(data=config_page_query_result)

@metataskController.get(
    '/tree', response_model=PageResponseModel, dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:list'))]
)
async def get_data_source_tree( request: Request,):
    # 获取分页数据

    data_tree_result = await MetataskService.get_data_source_tree( request)
    logger.info('获取成功')

    return ResponseUtil.success(rows=data_tree_result)
@metataskController.get(
    '/sourceall', response_model=PageResponseModel, dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:list'))]
)
async def get_data_source_all( request: Request,):
    # 获取分页数据

    data_source_result = await MetataskService.get_data_source_all( request)
    logger.info('获取成功')

    return ResponseUtil.success(data=data_source_result)
@metataskController.post('', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:add'))])
@ValidateFields(validate_model='add_metatask')
@Log(title='元数据任务', business_type=BusinessType.INSERT)
async def add_meta_metatask(
    request: Request,
    add_metatask: MetataskModel,
    query_db: AsyncSession = Depends(get_db),
    current_user: CurrentUserModel = Depends(LoginService.get_current_user),
):
    add_metatask.create_by = current_user.user.user_name
    add_metatask.create_time = datetime.now()
    add_metatask.update_by = current_user.user.user_name
    add_metatask.update_time = datetime.now()
    add_metatask_result = await MetataskService.add_metatask_services(request, query_db, add_metatask)
    logger.info(add_metatask_result.message)

    return ResponseUtil.success(msg=add_metatask_result.message)



@metataskController.put('', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:edit'))])
@ValidateFields(validate_model='edit_metatask')
@Log(title='元数据任务', business_type=BusinessType.UPDATE)
async def edit_meta_metatask(
    request: Request,
    edit_metatask: MetataskModel,
    query_db: AsyncSession = Depends(get_db),
    current_user: CurrentUserModel = Depends(LoginService.get_current_user),
):
    edit_metatask.update_by = current_user.user.user_name
    edit_metatask.update_time = datetime.now()
    edit_config_result = await MetataskService.edit_metatask_services(request, query_db, edit_metatask)
    logger.info(edit_config_result.message)

    return ResponseUtil.success(msg=edit_config_result.message)
# 元数据任务上下线
@metataskController.put('/upOrdown', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:up',"meta:metatask:down"))])
@Log(title='元数据任务', business_type=BusinessType.UPDATE)
async def up_or_down_meta_metatask(
    request: Request,
    DownOrUpdate: OperaMetataskModel,
    query_db: AsyncSession = Depends(get_db),
    current_user: CurrentUserModel = Depends(LoginService.get_current_user),
):

    edit_config_result = await MetataskService.up_or_down_metatask_services(request, query_db,current_user, DownOrUpdate.id,DownOrUpdate.type)
    logger.info(edit_config_result.message)

    return ResponseUtil.success(msg=edit_config_result.message)
# 元数据任务运行
@metataskController.put('/Run', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:run'))])
@Log(title='元数据任务', business_type=BusinessType.UPDATE)
async def run_meta_metatask(
    request: Request,
    process: ProcessDefinition,
    query_db: AsyncSession = Depends(get_db),
   
):

    edit_config_result = await MetataskService.run_metatask_services(request, query_db, process)

    return ResponseUtil.success(msg=edit_config_result)
# 元数据任务调度
@metataskController.put('/DS', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:ds'))])
@Log(title='元数据任务', business_type=BusinessType.UPDATE)
async def DS_meta_metatask(
    request: Request,
    process: ParmScheduleVo,
    query_db: AsyncSession = Depends(get_db),
   
):

    edit_config_result = await MetataskService.ds_metatask_services(request, query_db, process)

    return ResponseUtil.success(msg=edit_config_result)

# 元数据任务日志
@metataskController.get('/logs', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:up',"meta:metatask:down"))])
@Log(title='元数据任务', business_type=BusinessType.UPDATE)
async def get_metatask_logs(
    request: Request,
    id: str,
    query_db: AsyncSession = Depends(get_db),
    current_user: CurrentUserModel = Depends(LoginService.get_current_user),
):

    edit_config_result = await MetataskService.get_metatask_logs_services(request, query_db,current_user, id)
    logger.info(edit_config_result.message)

    return ResponseUtil.success(msg=edit_config_result.message)

@metataskController.delete('/{metatask_ids}/{ds_ids}', dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:delete'))])
@Log(title='元数据任务', business_type=BusinessType.DELETE)
async def delete_system_config(request: Request, metatask_ids: str,ds_ids:str, query_db: AsyncSession = Depends(get_db)):
    delete_config = DeleteMetataskModel(metatask_ids=metatask_ids,ds_ids=ds_ids)
    delete_config_result = await MetataskService.delete_metatask_services(request, query_db, delete_config)
    logger.info(delete_config_result.message)

    return ResponseUtil.success(msg=delete_config_result.message)



@metataskController.get(
    '/{metatask_id}', response_model=MetataskModel, dependencies=[Depends(CheckUserInterfaceAuth('meta:metatask:query'))]
)
async def query_detail_system_config(request: Request, metatask_id: int, query_db: AsyncSession = Depends(get_db)):
    metatask_detail_result = await MetataskService.metatask_detail_services(query_db, metatask_id)
    logger.info(f'获取config_id为{metatask_id}的信息成功')

    return ResponseUtil.success(data=metatask_detail_result)