import os
import argparse
import sys
import requests
import json
import subprocess
from datetime import datetime
# 定义一个装饰器,在方法执行异常时跳过执行
def skip_on_exception(func):
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
print(f"Exception occurred in {func.__name__}: {str(e)}")
# 跳过当前方法,执行下一个方法
pass
return wrapper
# 定义方法
@skip_on_exception
def excute_mysql_task():
#执行mysql2las任务和las2mysql任务
print("执行了mysql方法")
@skip_on_exception
def excute_bytehouse_ce_task():
print("执行了bytehouse_ce方法")
@skip_on_exception
def excute_es_task():
print("执行了es方法")
# 定义一个可以循环执行方法的嵌套方法,用来同时执行多个方法
def execute_dts_task(datasource_list):
for datasource in datasource_list:
print(datasource)
task = 'excute_'+datasource +'_task'
eval(task)
if __name__=='__main__':
#创建了一个参数解析器对象parser,使用formatter_class 参数设置默认帮助格式
parser=argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
#向参数解析器中,添加命令行参数,'--datasource' 是参数的名称,--表示是个选项;nargs='+' 表示该参数可以接受一个或多个值;default='mysql' 表示如果没有提供该参数,则使用默认值 'mysql';
parser.add_argument('--datasource',nargs='+',default='mysql',help='specify the datasource mysql bytehouse_ce es')
#解析命令行参数,并将结果存储在args里,args.datasource 是用户提供的数据源列表
args =parser.parse_args()
datasource_list = args.datasource
execute_dts_task(datasource_list)
print('DTS任务执行完成')
以上代码编辑完成后,在terminal里执行?python3 init_env.py --help,得到执行python3 init_env.py --datasource mysql 得到
执行python3 init_env.py --datasource mysql bytehouse_ce es 得到