【代码运行服务】在线编译

接口平台支持编写python代码,然后在用例或接口调试中以方法名称的方式中调用执行 @{fun_name()}@

但是存在的问题在于之前并没有在线调试的功能,用户在本地调试通过后拷贝代码到接口平台web端保存,执行用例时提示保存,再通过报错信息调试,这样既麻烦也不好排查问题
再者还有一个严重的问题在于,这些代码是直接嵌入到接口平台的代码中执行,有极大的潜在风险

现准备将在线调试的功能做为单独的服务,执行调试运行和调用执行

调试运行模块的代码

import os,sys,subprocess,tempfile,time
#tempfile创建临时文件夹
TempFile = tempfile.mkdtemp(suffix='_test', prefix='python_')
# 文件名
FileNum = int(time.time()*1000)
# python编译器位置
EXEC = sys.executable

#获取python版本
def get_version():
    v = sys.version_info
    version = "python %s.%s" %(v.major,v.minor)
    return version

# 获得py文件名
def get_pyname():
    global FileNum
    return 'test_%d' % FileNum

# 接收代码写入文件
def write_file(pyname, code):
    fpath = os.path.join(TempFile, '%s.py' % pyname)
    with open(fpath, 'w', encoding='utf-8') as f:
        f.write(code)
    print('file path: %s' % fpath)
    return fpath

# 编码
def decode(s):
    try:
        return s.decode('utf-8')
    except UnicodeDecodeError:
        return s.decode('gbk')

# 主执行函数
def main(code):
    r = dict()
    r["version"] = get_version()
    pyname = get_pyname()
    fpath = write_file(pyname, code)
    try:
        # subprocess.check_output 是父进程等待子进程完成,返回子进程向标准输出的输出结果
        # stderr是标准输出的类型
        outdata = decode(subprocess.check_output([EXEC, fpath], stderr=subprocess.STDOUT, timeout=5))
    except subprocess.CalledProcessError as e:
        # e.output是错误信息标准输出
        # 错误返回的数据
        r["code"] = 'Error'
        r["output"] = decode(e.output)
        return r
    else:
        # 成功返回的数据
        r['output'] = outdata
        r["code"]="Success"
        return r
    finally:
        # 删除文件(其实不用删除临时文件会自动删除)
        try:
            os.remove(fpath)
        except Exception as e:
            exit(1)

if __name__ == '__main__':
     code ="""print("你好")"""
     print(main(code))

其实调试运行的代码并不复杂,主要是subprocess.check_output([EXEC, fpath], stderr=subprocess.STDOUT, timeout=5)

subprocess.check_output函数可以执行一条sh命令,并返回命令的输出内容,用法如下:

output = subprocess.check_output(["python3", "xx.py"], shell = False)

该函数两个参数第一个表示命令内容,因为中间有空格所以用中括号这种形式,同时制定shell=False表示命令分开写了。而该命令执行后的输出内容会返回给output变量
timeout 设置执行的超时时间,单位为秒

让我们看看subprocess.check_output的使用

import shlex, subprocess
command_line = "echo 'hello' "
args = shlex.split(command_line)
print(args)
try:
    p = subprocess.check_output(args,stderr=subprocess.STDOUT,timeout=5)
    print(p)
except subprocess.TimeoutExpired as time_e:
    print(time_e)
except subprocess.CalledProcessError as call_e:
    print(e.output.decode(encoding="utf-8"))

1.timeout参数不能和shell=True一起使用,不然就算是时间到了,还是会继续执行,等执行结束以后才会抛出subprocess.TimeoutExpired异常,timeout的单位是秒。

2.check_output返回的是子程序的执行结果(上述demo返回的就应该是helo),也是unicode编码,如果程序执行报错的话,会直接抛出异常CalledProcessError,并且异常当中会有output属性,该属性为unicode编码的,要当字符串使用的时候需要转码,如e.output.decode(encoding=“utf-8”)

3.想要执行的命令command_line不能有类似<的重定向符号,会报错

原文地址:https://www.cnblogs.com/guanhuohuo/p/12533575.html