You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在非全局作用域执行动态函数以转换Pandas DataFrame?

问题

我正在基于Pandas开发一款简易报表工具,希望报表开发者能够通过函数转换DataFrame。具体流程为:开发者在UI界面创建数据库连接、SQL查询语句,以及若干以字符串形式编写的Python函数,后续这些函数将被执行。

我的类结构如下:

class ReportSQLDataset:
    def __init__(self, connection, sql, transforming_functions:list):
        self.connection = connection
        self.sql = sql
        self.dataframe = None
        self.transforming_functions = transforming_functions
        
    def load_dataframe(self):
        self.dataframe = pd.read_sql(self.sql, self.connection)

    def run_transforming_functions(self):
       for code in self.transforming_functions:
           # Here I want execute code from strings provided by user

转换函数示例:

def some_transformation(df):
    # Do something
    return df

请问有什么方法可以实现这一需求,且无需将这些函数放入全局作用域?

解决方案

方法1:用exec在局部命名空间执行函数

可以通过exec将用户提供的函数字符串执行在独立的局部命名空间中,提取函数后执行,完全不会污染全局作用域。

修改run_transforming_functions方法:

def run_transforming_functions(self):
    if self.dataframe is None:
        self.load_dataframe()
    df = self.dataframe
    for code in self.transforming_functions:
        # 创建独立的局部命名空间字典
        local_namespace = {}
        # 执行用户代码,函数会被存入local_namespace
        exec(code, {}, local_namespace)
        # 提取命名空间中的函数(假设每个字符串只定义一个函数)
        transform_func = next(iter(local_namespace.values()))
        # 执行转换并更新DataFrame
        df = transform_func(df)
    self.dataframe = df

关键说明:

  • 每次循环新建local_namespace,保证多个转换函数的命名空间完全隔离,互不干扰。
  • exec的第二个参数传入空字典,限制用户代码对全局作用域的访问,降低风险。
  • 如果约定了函数名(比如要求所有转换函数都叫transform_df),可以直接通过local_namespace['transform_df']提取,更可靠。

方法2:用ast模块解析并验证代码(更安全)

如果需要对用户代码做语法校验(比如确保只定义一个函数、禁止危险操作),可以用ast模块先解析代码,再执行。

示例代码:

import ast

def run_transforming_functions(self):
    if self.dataframe is None:
        self.load_dataframe()
    df = self.dataframe
    for code in self.transforming_functions:
        # 解析代码为抽象语法树(AST)
        code_ast = ast.parse(code)
        # 校验代码结构:只能包含一个函数定义
        if len(code_ast.body) != 1 or not isinstance(code_ast.body[0], ast.FunctionDef):
            raise ValueError("每个转换代码必须仅包含一个函数定义")
        # 编译AST为可执行对象
        compiled_code = compile(code_ast, filename="<user_transform_code>", mode="exec")
        # 执行编译后的代码到局部命名空间
        local_namespace = {}
        exec(compiled_code, {}, local_namespace)
        # 通过函数名提取函数并执行转换
        func_name = code_ast.body[0].name
        df = local_namespace[func_name](df)
    self.dataframe = df

关键说明:

  • 可以扩展AST校验逻辑,比如限制函数只能接受df作为参数,禁止函数体中出现导入危险模块、修改全局变量等操作。
  • 相比直接用exec,这种方式能提前拦截不符合要求的代码,提升安全性。

重要安全提示

  • 用户输入的代码存在极高安全风险,可能执行删除文件、窃取数据等恶意操作,必须在隔离环境(如沙箱、Docker容器)中运行。
  • 可以使用RestrictedPython库进一步限制代码权限,比如禁止访问文件系统、网络、系统命令等。
  • 永远不要在生产环境直接执行未经过滤的用户代码,必须做严格的权限控制和语法校验。

内容的提问来源于stack exchange,提问作者Vitaliy

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.07 06:45:28