from fastapi import Request,Depends from sqlalchemy.ext.asyncio import AsyncSession from module_admin.entity.vo.metatask_vo import MetataskQueryModel, MetataskModel, DeleteMetataskModel from module_admin.dao.metatask_dao import MetataskDao from utils.page_util import PageResponseModel # from module_admin.entity.vo.dataSource_vo import DataSource,Datasouceall,AlertGroups,Environment,WorkerGroup,ProcessDefinition,ParmScheduleVo,ParmSchedule,ProcessInstancePage from module_admin.entity.vo.dataSource_vo import * from module_admin.entity.vo.metaprocessconfig_vo import MetaprocessconfigQueryModel,MetaprocessconfigModel from config.constant import CommonConstant from exceptions.exception import ServiceException from module_admin.entity.vo.common_vo import CrudResponseModel from utils.common_util import CamelCaseUtil from module_admin.entity.vo.user_vo import CurrentUserModel from datetime import datetime import requests import json import re from config.enums import RedisInitKeyConfig from module_admin.service.metaprocessconfig_service import MetaprocessconfigService from config.env import AppConfig class MetataskService: """ 参数配置管理模块服务层 """ @classmethod async def get_metatask_list_services( cls, query_db: AsyncSession, query_object: MetataskQueryModel, is_page: bool = False ): """ 获取参数配置列表信息service :param query_db: orm对象 :param query_object: 查询参数对象 :return: 参数配置列表信息对象 """ metatask_list_result = await MetataskDao.get_metatask_list(query_db, query_object, is_page) return metatask_list_result @classmethod async def get_data_source_tree(cls,request: Request, current_user: CurrentUserModel): url = f'{AppConfig.ds_server_url}/dolphinscheduler/datasources?pageNo=1&pageSize=100' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password,} response = requests.get(url, headers=headers) if response.reason == 'OK': response_text = response.text data = json.loads(response_text) total_list = data["data"]["totalList"] # 解析 connectionParams 字符串为字典 for item in total_list: item["connectionParams"] = json.loads(item["connectionParams"]) # 使用 Pydantic 创建 DataSource 对象列表 data_sources = [DataSource(**item) for item in total_list] return data_sources else: return {'error': f'Request failed with status code {response.status_code}'} @classmethod async def get_data_source_all(cls,request: Request,current_user: CurrentUserModel): # Worker分组 url1 = f'{AppConfig.ds_server_url}/dolphinscheduler/worker-groups/all' # 警告组 url2= f'{AppConfig.ds_server_url}/dolphinscheduler/alert-groups/list' # 工作环境 url3 = f'{AppConfig.ds_server_url}/dolphinscheduler/environment/query-environment-list' dataspurceVo=Datasouceall() headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} response1 = requests.get(url1, headers=headers) response2 = requests.get(url2, headers=headers) response3 = requests.get(url3, headers=headers) if response1.reason == 'OK': response_text = response1.text data = json.loads(response_text) total_list = data["data"] # 使用 Pydantic 创建 DataSource 对象列表 data_sources = [WorkerGroup(name=item) for item in total_list] dataspurceVo.workerGroup=data_sources if response2.reason == 'OK': response_text = response2.text data = json.loads(response_text) total_list = data["data"] # 使用 Pydantic 创建 DataSource 对象列表 alertGroups = [AlertGroups(**item) for item in total_list] dataspurceVo.alertGroups=alertGroups if response3.reason == 'OK': response_text = response3.text data = json.loads(response_text) total_list = data["data"] data_sources = [Environment(**item) for item in total_list] dataspurceVo.environment=data_sources return dataspurceVo @classmethod async def check_metatask_name_unique_services(cls, query_db: AsyncSession, page_object: MetataskModel): """ 校验参数键名是否唯一service :param query_db: orm对象 :param page_object: 参数配置对象 :return: 校验结果 """ metatask_id = -1 if page_object.metatask_id is None else page_object.metatask_id metatask = await MetataskDao.get_metatask_detail_by_info( query_db, MetataskModel(metataskName=page_object.metatask_name) ) if metatask and metatask.metatask_id != metatask_id: return CommonConstant.NOT_UNIQUE return CommonConstant.UNIQUE @classmethod async def add_metatask_services(cls, request: Request, query_db: AsyncSession, page_object: MetataskModel,current_user: CurrentUserModel): """ 新增参数配置信息service :param request: Request对象 :param query_db: orm对象 :param page_object: 新增参数配置对象 :return: 新增参数配置校验结果 """ if not await cls.check_metatask_name_unique_services(query_db, page_object): raise ServiceException(message=f'新增元数据任务{page_object.metatask_name}失败，任务名已存在') else: try: # 查询模板 processconfig = MetaprocessconfigQueryModel() processconfig.ac_target=page_object.ac_target processconfig.db_type=page_object.dbCode processConfigList =await MetaprocessconfigService.get_metaprocessconfig_list_services(query_db,processconfig,False) message= await cls.mysql_process_defind_change_add(request,processConfigList,page_object,current_user) if "成功" not in message: await query_db.rollback() raise ServiceException(message=f'新增元数据任务{page_object.metatask_name}失败，dolphinscheduler创建失败') await MetataskDao.add_metatask_dao(query_db, page_object) await query_db.commit() return CrudResponseModel(is_success=True, message=message) except Exception as e: await query_db.rollback() raise e @classmethod async def mysql_process_defind_change_add(cls,request: Request, processConfigList: list[MetaprocessconfigModel],page_object:MetataskModel,current_user: CurrentUserModel): """ mysql类型 """ projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url = f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/task-definition/gen-task-codes?genNum=5' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} # 新增接口 url2=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers2 = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} response = requests.get(url, headers=headers) if response.reason == 'OK': intdsids=[] message='' dstypes=[] response_text = response.text data = json.loads(response_text) code_list = data["data"] str_list = list(map(str, code_list)) for config in processConfigList: # mysql表字段 if config.ac_target=='0': modified_json_str = config.taskDefinitionJson.replace("16199683466336", str_list[0]).replace("16199683466337", str_list[1]).replace("16199683466338", str_list[2]).replace("'dash_test_w'", f"'{page_object.dbSName}'").replace("'mysql_conn'", f"'{page_object.dbRName}'") modified_json_str2=config.taskRelationJson.replace("16199683466336", str_list[0]).replace("16199683466337", str_list[1]).replace("16199683466338", str_list[2]) modified_json_str3=config.locations.replace("16199683466336", str_list[0]).replace("16199683466337", str_list[1]).replace("16199683466338", str_list[2]) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description=page_object.remark, # 替换工作流备注 locations=modified_json_str3,# 替换locations name =page_object.metatask_name+"-表字段采集" ,# 替换工作流名称 timeout=config.timeout, globalParams =config.globalParams , tenantCode =config.tenantCode , taskRelationJson =modified_json_str2,# 替换taskRelationJson executionType =config.executionType , releaseState=config.releaseState ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_post0 = requests.post(url2, headers=headers2,data=form_data) # text= '{"code":0,"msg":"success","data":{"id":null,"code":16244510692320,"name":"测试2250-表字段采集","version":0,"releaseState":null,"projectCode":15094503753824,"description":"111","globalParams":"[]","globalParamList":null,"globalParamMap":{},"createTime":"2025-01-08 12:50:39","updateTime":"2025-01-08 12:50:39","flag":"YES","userId":1,"userName":null,"projectName":null,"locations":"[{\\"taskCode\\":16244510678624,\\"x\\":334,\\"y\\":265},{\\"taskCode\\":16244510678625,\\"x\\":334,\\"y\\":390},{\\"taskCode\\":16244510678626,\\"x\\":687,\\"y\\":335}]","scheduleReleaseState":null,"timeout":0,"tenantId":1,"tenantCode":null,"modifyBy":null,"warningGroupId":0,"executionType":"PARALLEL"},"failed":false,"success":true}' text= response_post0.text responsJson = json.loads(text) if responsJson['msg'] == 'success': intdsids.append(responsJson['data']['code']) dstypes.append('0') if message: message += ", " message += page_object.metatask_name + "-表字段采集新增成功" else: if message: message += ", " message += page_object.metatask_name + "-表字段采集新增失败" if config.ac_target=='1': modified_json_str = config.taskDefinitionJson.replace("16286410625888", str_list[0]).replace("16286410625889", str_list[1]).replace("16286410625890", str_list[2]).replace("16286410625891", str_list[3]).replace("'dash_test_w'", f"'{page_object.dbSName}'").replace("'mysql_conn'", f"'{page_object.dbRName}'").replace("mysql_conn dash_test_w", f"{page_object.dbRName} {page_object.dbSName}") modified_json_str2=config.taskRelationJson.replace("16286410625888", str_list[0]).replace("16286410625889", str_list[1]).replace("16286410625890", str_list[2]).replace("16286410625891", str_list[3]) modified_json_str3=config.locations.replace("16286410625888", str_list[0]).replace("16286410625889", str_list[1]).replace("16286410625890", str_list[2]).replace("16286410625891", str_list[3]) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description=page_object.remark, # 替换工作流备注 locations=modified_json_str3,# 替换locations name =page_object.metatask_name+"-存储过程采集" ,# 替换工作流名称 timeout=config.timeout, globalParams =config.globalParams , tenantCode =config.tenantCode , taskRelationJson =modified_json_str2,# 替换taskRelationJson executionType =config.executionType , releaseState=config.releaseState ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_post0 = requests.post(url2, headers=headers2,data=form_data) #text= '{"code":0,"msg":"success","data":{"id":null,"code":16244510692320,"name":"测试2250-表字段采集","version":0,"releaseState":null,"projectCode":15094503753824,"description":"111","globalParams":"[]","globalParamList":null,"globalParamMap":{},"createTime":"2025-01-08 12:50:39","updateTime":"2025-01-08 12:50:39","flag":"YES","userId":1,"userName":null,"projectName":null,"locations":"[{\\"taskCode\\":16244510678624,\\"x\\":334,\\"y\\":265},{\\"taskCode\\":16244510678625,\\"x\\":334,\\"y\\":390},{\\"taskCode\\":16244510678626,\\"x\\":687,\\"y\\":335}]","scheduleReleaseState":null,"timeout":0,"tenantId":1,"tenantCode":null,"modifyBy":null,"warningGroupId":0,"executionType":"PARALLEL"},"failed":false,"success":true}' text= response_post0.text responsJson = json.loads(text) if responsJson['msg'] == 'success': dstypes.append('1') intdsids.append(responsJson['data']['code']) if message: message += ", " message += page_object.metatask_name + "-存储过程采集新增成功" else: if message: message += ", " message += page_object.metatask_name + "-存储过程采集新增失败" if len(intdsids)>0: page_object.ds_ids=','.join([str(i) for i in intdsids]) page_object.ds_types=','.join([str(i) for i in dstypes]) return message @classmethod async def mysql_process_defind_change_update(cls,request: Request,page_object:MetataskModel,metatask_old:MetataskModel,current_user: CurrentUserModel): """ mysql类型 """ projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') dstypes=page_object.ds_types.split(",") dsids=page_object.ds_ids.split(",") result_list = [{'dstype': t, 'dsid': i} for t, i in zip(dstypes, dsids)] message='' # 查询接口 url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, } headers2 = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} for config in result_list: # mysql表字段 if config['dstype']=='0' : response = requests.get(f"{url}/{config['dsid']}", headers=headers) text= response.text responsJson = json.loads(text) if responsJson['msg'] == 'success': modified_json_str = json.dumps(responsJson['data']['taskDefinitionList'], ensure_ascii=False, indent=0).replace( f"'{metatask_old.dbSName}'", f"'{page_object.dbSName}'") getTaskRelationList=responsJson['data']['processTaskRelationList'] putTaskRelationList=[] for item in getTaskRelationList: new_item = { "name": item['name'], "preTaskCode":item['preTaskCode'] , "preTaskVersion":item['preTaskVersion'] , "postTaskCode":item['postTaskCode'] , "conditionType":item['conditionType'] , "conditionParams":item['conditionParams'] } putTaskRelationList.append(new_item) modified_json_str2= json.dumps(putTaskRelationList, ensure_ascii=False, indent=0) modified_json_str2=re.sub(r'\s+', '', modified_json_str2) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description=page_object.remark, # 替换工作流备注 locations=responsJson['data']['processDefinition']['locations'] ,# 替换locations name =page_object.metatask_name+"-表字段采集" ,# 替换工作流名称 tenantCode =responsJson['data']['processDefinition']['tenantCode'] , taskRelationJson =modified_json_str2,# 替换taskRelationJson ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_put0 = requests.put(f"{url}/{config['dsid']}", headers=headers2,data=form_data) putText= response_put0.text responsPutJson=json.loads(putText) if responsPutJson['msg'] == 'success': if message: message += ", " message += page_object.metatask_name + "-表字段采集修改成功" else: if message: message += ", " message += page_object.metatask_name + "-表字段采集修改失败" if config['dstype']=='1': response = requests.get(f"{url}/{config['dsid']}", headers=headers) text= response.text responsJson = json.loads(text) if responsJson['msg'] == 'success': modified_json_str = json.dumps(responsJson['data']['taskDefinitionList'], ensure_ascii=False, indent=4).replace( f"'{metatask_old.dbSName}'", f"'{page_object.dbSName}'") getTaskRelationList=responsJson['data']['processTaskRelationList'] putTaskRelationList=[] for item in getTaskRelationList: new_item = { "name": item['name'], "preTaskCode":item['preTaskCode'] , "preTaskVersion":item['preTaskVersion'] , "postTaskCode":item['postTaskCode'] , "conditionType":item['conditionType'] , "conditionParams":item['conditionParams'] } putTaskRelationList.append(new_item) modified_json_str2= json.dumps(putTaskRelationList, ensure_ascii=False, indent=4) modified_json_str2=re.sub(r'\s+', '', modified_json_str2) metaprocessconfig_dict = MetaprocessconfigModel(taskDefinitionJson=modified_json_str,# 替换taskDefinitionJson description=page_object.remark, # 替换工作流备注 locations=responsJson['data']['processDefinition']['locations'] ,# 替换locations name =page_object.metatask_name+"-存储过程采集" ,# 替换工作流名称 tenantCode =responsJson['data']['processDefinition']['tenantCode'] , taskRelationJson =modified_json_str2,# 替换taskRelationJson ).model_dump(exclude_unset=True, by_alias=True) form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response_put0 = requests.put(f"{url}/{config['dsid']}", headers=headers2,data=form_data) putText= response_put0.text responsPutJson=json.loads(putText) if responsPutJson['msg'] == 'success': if message: message += ", " message += page_object.metatask_name + "-存储过程采集修改成功" else: if message: message += ", " message += page_object.metatask_name + "-存储过程采集修改失败" return message @classmethod async def edit_metatask_services(cls, request: Request, query_db: AsyncSession, page_object: MetataskModel,current_user: CurrentUserModel): """ 编辑参数配置信息service :param request: Request对象 :param query_db: orm对象 :param page_object: 编辑参数配置对象 :return: 编辑参数配置校验结果 """ edit_metatask = page_object.model_dump(exclude_unset=True) metatask_info = await cls.metatask_detail_services(query_db, page_object.metatask_id) if metatask_info.metatask_id: if not await cls.check_metatask_name_unique_services(query_db, page_object): raise ServiceException(message=f'修改任务{page_object.metatask_name}失败，任务名称已存在') else: try: metatask_old = await cls.metatask_detail_services(query_db, metatask_info.metatask_id) message= await cls.mysql_process_defind_change_update(request,page_object,metatask_old,current_user) if "成功" not in message: await query_db.rollback() raise ServiceException(message=f'修改元数据任务{page_object.metatask_name}失败，dolphinscheduler修改失败') await MetataskDao.edit_metatask_dao(query_db, edit_metatask) await query_db.commit() return CrudResponseModel(is_success=True, message='更新成功') except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='更新失败') @classmethod async def up_or_down_metatask_services( cls, request: Request, query_db: AsyncSession, current_user: CurrentUserModel, id: str, type: str ): metatask_info = await cls.metatask_detail_services(query_db, id) metatask_info.update_by = current_user.user.user_name metatask_info.update_time = datetime.now() type_str: str projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') dstypes=metatask_info.ds_types.split(",") dsids=metatask_info.ds_ids.split(",") result_list = [{'dstype': t, 'dsid': i} for t, i in zip(dstypes, dsids)] message='' # 查询接口 url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} for config in result_list: # mysql表字段 if config['dstype']=='0' : metaprocessconfig_dict = { 'name': metatask_info.metatask_name+'-表字段采集', 'releaseState': type } form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response = requests.post(f"{url}/{config['dsid']}/release", headers=headers,data=form_data) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: message='成功！' else: raise ServiceException(message='失败'+responsJson['msg']) # mysql表字段 if config['dstype']=='1' : metaprocessconfig_dict = { 'name': metatask_info.metatask_name+'-存储过程采集', 'releaseState': type } form_data = {key: str(value) for key, value in metaprocessconfig_dict.items()} response = requests.post(f"{url}/{config['dsid']}/release", headers=headers,data=form_data) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: message='成功！' else: raise ServiceException(message='失败'+responsJson['msg']) if type == 'OFFLINE': # 下线 type_str = '下线' metatask_info.status = 'OFFLINE' else: # 上线 type_str = '上线' metatask_info.status = 'ONLINE' edit_metatask = metatask_info.model_dump(exclude_unset=True) try: await MetataskDao.edit_metatask_dao(query_db, edit_metatask) await query_db.commit() return CrudResponseModel(is_success=True, message=message) except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='更新失败') @classmethod async def run_metatask_services( cls, request: Request, query_db: AsyncSession, process: ProcessDefinition,current_user: CurrentUserModel ): process.failureStrategy='CONTINUE' process.warningType='NONE' process.execType='START_PROCESS' process.taskDependType='TASK_POST' process.complementDependentMode ='OFF_MODE' process.runMode='RUN_MODE_SERIAL' process.processInstancePriority='MEDIUM' process.dryRun=0 process.scheduleTime="{complementStartDate:'2025-01-12 00:00:00',complementEndDate:'2025-01-12 00:00:00'}" projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/executors/start-process-instance' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} # form_data = {key: str(value) for key, value in process.__dict__.items()} form_data = {key: value for key, value in process.__dict__.items()} response = requests.post(url, headers=headers,data=form_data) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: return "运行成功！" else: raise ServiceException(message='运行失败！') @classmethod async def ds_metatask_services( cls, request: Request, query_db: AsyncSession, process: ParmScheduleVo,current_user: CurrentUserModel ): parm =ParmSchedule( ) parm.failureStrategy='CONTINUE' parm.warningType='NONE' parm.warningGroupId=process.warningGroupId parm.workerGroup='TASK_POST' parm.processDefinitionCode =process.processDefinitionCode parm.environmentCode=process.environmentCode parm.processInstancePriority='MEDIUM' parm.schedule = ( '{"startTime":"' + process.beginTime.strftime('%Y-%m-%d %H:%M:%S') + '", "endTime":"' + process.endTime.strftime('%Y-%m-%d %H:%M:%S') + '", "crontab":"' + process.crontab + '", "timezoneId":"Asia/Shanghai"}') projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/schedules' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} # form_data = {key: str(value) for key, value in process.__dict__.items()} form_data = {key: value for key, value in parm.__dict__.items()} response = requests.post(url, headers=headers,data=form_data) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: return "运行成功！" else: raise ServiceException(message='运行失败！') @classmethod async def get_metatask_logs_services( cls, request: Request, current_user: CurrentUserModel, query_db: AsyncSession, id: str ): metatask_info = await cls.metatask_detail_services(query_db, id) metatask_info.update_by = current_user.user.user_name metatask_info.update_time = datetime.now() # 运行中 metatask_info = '3' edit_metatask = metatask_info.model_dump(exclude_unset=True) try: await MetataskDao.edit_metatask_dao(query_db, edit_metatask) await query_db.commit() return CrudResponseModel(is_success=True, message=metatask_info.metatask_name + '任务：' '日志获取成功') except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='更新失败') @classmethod async def delete_metatask_services(cls, request: Request, query_db: AsyncSession, page_object: DeleteMetataskModel,current_user: CurrentUserModel): """ 删除参数配置信息service :param request: Request对象 :param query_db: orm对象 :param page_object: 删除参数配置对象 :return: 删除参数配置校验结果 """ if page_object.metatask_ids and page_object.ds_ids: metatask_id_list = page_object.metatask_ids.split(',') try: projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') # 查询接口 url=f'{AppConfig.ds_server_url}/dolphinscheduler/projects/'+projectCode+'/process-definition/batch-delete' form_data={'codes':page_object.ds_ids} headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password, 'Content-Type': 'application/x-www-form-urlencoded'} response = requests.post(url, headers=headers,data=form_data) text= response.text responsJson = json.loads(text) if responsJson['success'] is True: for metatask_id in metatask_id_list: await MetataskDao.delete_metatask_dao(query_db, MetataskModel(metataskId=int(metatask_id))) await query_db.commit() return CrudResponseModel(is_success=True, message='删除成功') else : raise ServiceException(message='ds删除失败') except Exception as e: await query_db.rollback() raise e else: raise ServiceException(message='传入参数配置id为空') @classmethod async def metatask_detail_services(cls, query_db: AsyncSession, metatask_id: int): """ 获取参数配置详细信息service :param query_db: orm对象 :param metatask_id: 参数配置id :return: 参数配置id对应的信息 """ metatask = await MetataskDao.get_metatask_detail_by_id(query_db, metatask_id=metatask_id) if metatask: result = MetataskModel(**CamelCaseUtil.transform_result(metatask)) else: result = MetataskModel(**dict()) return result @classmethod async def get_process_instances_services( cls, request: Request, query_object: ProcessInstancePage,current_user: CurrentUserModel ): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url = f'{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/process-instances?pageNo={query_object.page_num}&pageSize={query_object.page_size}&searchVal={query_object.searchVal}' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} response = requests.get(url, headers=headers) try: if response.reason == 'OK': response_text = response.text data = json.loads(response_text) total_list = data["data"]["totalList"] # data_sources = [ProcessInstance(**item) for item in total_list] pageData = PageResponseModel(rows=total_list,total=data["data"]["total"]) return pageData else: return {'error': f'Request failed with status code {response.status_code}'} except Exception as e: raise e @classmethod async def get_task_nodes_services( cls, request: Request,id:int,current_user: CurrentUserModel ): projectCode = await request.app.state.redis.get(f'{RedisInitKeyConfig.SYS_CONFIG.key}:sys.ds.projectcode') url = f'{AppConfig.ds_server_url}/dolphinscheduler/projects/{projectCode}/process-instances/{id}/tasks' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} response = requests.get(url, headers=headers) try: response = requests.get(url, headers=headers) if response.reason == 'OK': response_text = response.text data = json.loads(response_text) total_list = data["data"]["taskList"] data_sources = [TaskNode(**item) for item in total_list] return data_sources else: return {'error': f'Request failed with status code {response.status_code}'} except Exception as e: raise e @classmethod async def get_log_details_services( cls, request: Request,id:int,current_user: CurrentUserModel ): url = f'{AppConfig.ds_server_url}/dolphinscheduler/log/detail?taskInstanceId={id}&limit=1000&skipLineNum=0' headers = {'dashUserName': current_user.user.user_name, 'dashPassword': current_user.user.password} response = requests.get(url, headers=headers) try: response = requests.get(url, headers=headers) if response.reason == 'OK': response_text = response.text data = json.loads(response_text) logMessage = data["data"]["message"] return logMessage else: return {'error': f'Request failed with status code {response.status_code}'} except Exception as e: raise e