api-datamanager/Modules/CodeExecutor/PythonCodeExecutor.py

53 lines
1.6 KiB
Python

from Utils.ErrorUtil import JustThrowError
class PythonCodeExecutor(object):
def __init__(self, func_name, func_text, params):
self.func_name = ExecutorUtils.check_func_name(func_name)
self.func_text = ExecutorUtils.check_code(func_text)
self.params = ExecutorUtils.check_params(params)
self.result = self.impl()
def impl(self):
try:
exec(self.func_text)
return eval("{func_name}({params})".format(func_name=self.func_name, params=self.params))
except Exception:
raise JustThrowError("代码执行失败,请检查传入参数是否符合要求")
class ExecutorUtils(object):
@staticmethod
def check_func_name(func_name):
if type(func_name) != str:
raise JustThrowError("方法名称不符合文本格式")
return func_name
@staticmethod
def check_code(code_text):
if type(code_text) != str:
raise JustThrowError("代码不符合文本格式")
# 禁止使用import、eval、exec
ban_words = ["import", "eval", "exec"]
for ban_word in ban_words:
if ban_word in code_text:
raise JustThrowError("代码中带有非法字符")
return code_text
@staticmethod
def check_params(input_params):
if type(input_params) != dict:
raise JustThrowError("传入参数不符合字典格式")
return_params = ""
for item in input_params.items():
return_params = return_params + ("{}={}, ".format(item[0], item[1]))
return_params = return_params[:-2]
return return_params