FastAPI中如何对同一数据元素应用多个处理方法

你要实现的是字符串的顺序处理流水线，核心用到Python「函数是一等公民」的特性，不需要逐行写30次函数调用，避免后续维护成本过高。

核心前置概念 #

• 函数可以作为对象存入列表、按顺序遍历，存入列表时不需要提前执行
• 提前约定处理链路的传参规则：是上一步的输出作为下一步的输入（适合字符串加工、转换场景），还是所有函数都接收原始字符串（适合多规则校验、日志打点场景）
• 提前约定中断规则：某一步处理失败/校验不通过时，是否直接终止后续流程返回结果

基础实现代码 #

先把所有自定义处理函数按你需要的执行顺序，整理成函数列表（注意只写函数名，不要加括号，加括号会直接执行函数），之后在接口里循环遍历执行即可：

from fastapi import FastAPI
app = FastAPI()

# 按顺序定义你的30个处理函数
def func1(data: str):
    # 示例：去除字符串首尾空白
    return data.strip()
def func2(data: str):
    # 示例：敏感词替换
    return data.replace("违禁内容", "***")
# func3 ~ func29 按你的实际需求定义
def func30(data: str):
    # 示例：最终结果封装
    return {"process_result": data}

# 严格按执行顺序排列处理函数
PROCESS_CHAIN = [
    func1,
    func2,
    func3,
    # 按顺序补全到func30
    func30
]

@app.get("/process/{data}")
def process(data: str):
    current_val = data
    # 依次执行所有处理函数
    for handler in PROCESS_CHAIN:
        current_val = handler(current_val)
    return current_val

这种写法后续要调整执行顺序、新增/删除处理步骤，只需要修改PROCESS_CHAIN列表即可，不需要改动接口主逻辑，可维护性远高于逐行硬写调用。

可选优化点 #

• 增加流程短路逻辑：如果某一步是校验类逻辑，校验不通过时直接终止流程返回错误，不需要执行后续步骤。比如约定处理函数返回False代表校验失败：

@app.get("/process/{data}")
def process(data: str):
    current_val = data
    for step, handler in enumerate(PROCESS_CHAIN, start=1):
        current_val = handler(current_val)
        if current_val is False:
            return {"code": 400, "msg": f"第{step}步处理{handler.__name__}校验不通过"}
    return {"code": 200, "result": current_val}

• 增加全局异常捕获：避免某一步处理函数抛出未捕获异常，导致接口直接返回500错误：

@app.get("/process/{data}")
def process(data: str):
    current_val = data
    try:
        for step, handler in enumerate(PROCESS_CHAIN, start=1):
            current_val = handler(current_val)
            if current_val is False:
                return {"code": 400, "msg": f"第{step}步处理{handler.__name__}校验不通过"}
    except Exception as e:
        return {"code": 500, "msg": f"第{step}步处理异常：{str(e)}"}
    return {"code": 200, "result": current_val}

• 如果你的处理逻辑不需要链式传递修改后的值，所有函数都只需要读取原始字符串做校验/统计，那循环时直接传入原始data即可，不需要更新current_val变量。

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