如何通过FastAPI接收用户规则并自动应用于Pandas DataFrame

我有一个Pandas DataFrame（以diamonds数据集为例），希望通过FastAPI、Python、Pandas实现接收用户自定义规则并将其应用到DataFrame中。当前示例代码是预设相关性判断规则，现需改为接收用户任意规则并执行。原有示例代码如下：

def rules(get_target_dir : utils.target_directory):
    global df
    col1 = get_target_dir.column1
    col2 = get_target_dir.column2
    target_directory = get_target_dir.target_directory
    df = utils.df
    corr = utils.df[col1].corr(df[col2])
    if corr > 0.90:
        print("highly correlated")
    else:
        print("not correlated")

实现思路与代码 #

要支持用户自定义规则，核心是安全接收并执行用户提供的规则逻辑，同时规避代码注入风险。以下是具体实现步骤：

1. 定义请求模型

用Pydantic模型统一接收用户传入的规则字符串及可选参数：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import pandas as pd
from typing import Optional

# 加载示例diamonds数据集
df = pd.read_csv("diamonds.csv")

app = FastAPI()

class RuleRequest(BaseModel):
    rule: str  # 用户自定义的规则代码片段
    columns: Optional[list[str]] = None  # 规则涉及的列名（可选）

2. 安全执行用户规则

通过限制可用变量范围，使用exec()执行用户规则，同时捕获异常返回错误信息：

@app.post("/apply-rule")
async def apply_rule(request: RuleRequest):
    try:
        # 仅暴露安全的变量给用户规则
        local_vars = {
            "df": df.copy(),  # 传入副本避免原数据被修改
            "pd": pd
        }
        
        # 执行用户提交的规则
        exec(request.rule, globals(), local_vars)
        
        # 提取规则执行结果（约定用户通过result变量返回结果）
        result = local_vars.get("result", "规则执行完成")
        return {"status": "success", "result": result}
    
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"规则执行失败: {str(e)}")

3. 用户规则示例

用户可传入符合Python语法的规则代码，示例如下：

• 自定义相关性判断规则：

col1 = 'carat'
col2 = 'price'
corr = df[col1].corr(df[col2])
result = "高度相关" if corr > 0.8 else "相关性较低"

• 数据筛选与统计规则：

high_value_df = df[(df['price'] > 5000) & (df['cut'] == 'Premium')]
result = f"高价优质钻石数量: {len(high_value_df)}, 平均克拉数: {high_value_df['carat'].mean():.2f}"

4. 关键注意事项

• 安全防护：生产环境需添加沙箱隔离、代码白名单、权限校验等措施，避免恶意代码注入
• 数据隔离：传入DataFrame副本，防止用户规则修改原始数据集
• 规则约定：要求用户通过result变量返回执行结果，保证接口响应格式统一

内容的提问来源于stack exchange，提问作者Abhishek Vashishth