如何解决exec执行代码后process函数未定义的NameError问题?
问题描述
我有一个包含Python函数的base64字符串,想要运行其中的代码。我的代码如下:
import base64 def apply_script(custom_script: str, spark_df=None): script_encoded = base64.b64decode(custom_script).decode('utf-8') print(script_encoded) exec(script_encoded) print("EXECUTE: ") process(spark_df) # name "process" imported from exec() function return None apply_script("aW1wb3J0IGpzb24KCgpkZWYgcHJvY2VzcyhkZik6CiAgICBwcmludCgiSGVsbG8gd29ybGQiKQogICAgcmV0dXJuIGRm")
运行后得到如下输出与错误:
python test_custom_script.py import json def process(df): print("Hello world") return df EXECUTE: Traceback (most recent call last): File "test_custom_script.py", line 13, in <module> apply_script("aW1wb3J0IGpzb24KCgpkZWYgcHJvY2VzcyhkZik6CiAgICBwcmludCgiSGVsbG8gd29ybGQiKQogICAgcmV0dXJuIGRm") File "test_custom_script.py", line 9, in apply_script process(spark_df) # name "process" imported from exec() function NameError: name 'process' is not defined
修复方案
问题根源是exec()执行代码时,默认不会将定义的函数注册到当前函数的局部作用域中。以下两种方式可以解决:
方式一:使用独立命名空间获取函数(推荐)
创建一个自定义字典作为exec()的局部命名空间,执行后从字典中取出process函数,避免污染当前作用域:
import base64 def apply_script(custom_script: str, spark_df=None): script_encoded = base64.b64decode(custom_script).decode('utf-8') print(script_encoded) # 初始化局部命名空间字典 local_ns = {} # 传入全局和局部命名空间执行脚本 exec(script_encoded, globals(), local_ns) print("EXECUTE: ") # 从命名空间中提取process函数并调用 process_func = local_ns['process'] process_func(spark_df) return None apply_script("aW1wb3J0IGpzb24KCgpkZWYgcHJvY2VzcyhkZik6CiAgICBwcmludCgiSGVsbG8gd29ybGQiKQogICAgcmV0dXJuIGRm")
方式二:直接注入当前函数的局部作用域
让exec()使用当前函数的局部作用域,这种方式简单但存在变量覆盖风险:
import base64 def apply_script(custom_script: str, spark_df=None): script_encoded = base64.b64decode(custom_script).decode('utf-8') print(script_encoded) # 传入当前函数的局部作用域执行脚本 exec(script_encoded, globals(), locals()) print("EXECUTE: ") process(spark_df) return None apply_script("aW1wb3J0IGpzb24KCgpkZWYgcHJvY2VzcyhkZik6CiAgICBwcmludCgiSGVsbG8gd29ybGQiKQogICAgcmV0dXJuIGRm")
内容的提问来源于stack exchange,提问作者mascai
相关产品推荐
相关产品推荐

