ETL主函数嵌套调用问题：如何传递内部函数返回值？

你的代码核心问题是内部函数调用时没有传递必要参数，导致数据无法在各步骤间流转，同时错误的参数传递也引发了后续操作的句柄无效错误。以下是修正思路和代码示例：

问题点拆解 #

• 调用extract()时未传入外部的url参数，导致爬取逻辑无法获取目标地址
• transform()和load()调用时未传入前一步的返回结果，数据链断裂
• 无需使用global，通过函数参数传递数据是更规范、无副作用的方式

修正后的代码 #

def etl(url):
    # 提取：接收URL，返回解析后的JSON数据
    def extract(input_url):
        # 替换为实际爬取+JSON解析逻辑
        import requests
        response = requests.get(input_url)
        response.raise_for_status()  # 主动抛出请求错误，提前排查问题
        return response.json()
    
    # 转换：接收原始JSON，返回整理后的字典
    def transform(raw_json):
        # 替换为实际数据转换逻辑
        transformed_data = {
            "target_field1": raw_json.get("source_field1"),
            "target_field2": raw_json.get("source_field2")
        }
        return transformed_data
    
    # 加载：接收整理后的数据，写入数据库
    def load(processed_data):
        # 替换为实际数据库写入逻辑（如SQL插入语句）
        print(f"成功加载数据到X数据库：{processed_data}")
    
    # 核心：按顺序调用函数，传递前一步的返回值
    raw_json_data = extract(url)
    cleaned_data = transform(raw_json_data)
    load(cleaned_data)

关键说明 #

1. 数据流转逻辑：每个步骤的输出直接作为下一个步骤的输入，形成extract → transform → load的完整链路
2. 错误前置处理：加入response.raise_for_status()可以在爬取阶段就捕获请求错误，避免无效数据流入后续步骤引发句柄错误
3. 职责分离：每个内部函数只负责单一任务，参数明确，便于单独测试和维护

内容的提问来源于stack exchange，提问作者Icharo-tb