调用独立函数处理文件时触发ValueError：已关闭文件的I/O操作

错误原因 #

你遇到的ValueError: I/O operation on closed file是因为生成器的惰性求值特性：load_from_csv里的with语句在返回_read_and_yield(reader)后就会立即关闭文件，但生成器并没有立即执行迭代操作——当你在外部循环调用生成器时，文件已经被关闭，自然无法读取数据。

核心解决思路 #

要保证文件在迭代生成器的全程处于打开状态，同时保持数据处理逻辑的独立性，推荐以下两种方案：

方案一：在上下文内转发生成器结果 #

修改load_from_csv，在with块内部通过yield from转发处理后的行数据，这样文件会在整个迭代过程中保持打开，同时_process_rows（原_read_and_yield）依然独立，可复用给其他文件类型。

import csv

class Loader():
    def load_from_csv(self, csv_path: str):
        with open(csv_path, mode='r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            # 在with上下文内迭代处理逻辑并转发结果
            yield from self._process_rows(reader)
    
    @classmethod
    def _process_rows(cls, row_iterator):
        for row in row_iterator:
            # 后续可在此扩展处理逻辑：类型转换、数据校验、字段过滤等
            yield row

# 使用示例
loader = Loader()
for row in loader.load_from_csv('sample.csv'):
    print(row)

方案二：扩展多文件类型支持 #

如果后续要支持Excel等其他格式，只需新增对应文件的读取方法，复用同一个_process_rows处理逻辑，只需保证传入的是统一格式的行迭代器即可：

import csv
import pandas as pd

class Loader():
    def load_from_csv(self, csv_path: str):
        with open(csv_path, mode='r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            yield from self._process_rows(reader)
    
    def load_from_excel(self, excel_path: str, sheet_name: str = 0):
        with pd.ExcelFile(excel_path) as xls:
            df = pd.read_excel(xls, sheet_name=sheet_name)
            # 将DataFrame转为与CSV reader兼容的字典迭代器
            yield from self._process_rows(df.to_dict('records'))
    
    @classmethod
    def _process_rows(cls, row_iterator):
        for row in row_iterator:
            # 示例扩展：将age字段转为整数
            if 'age' in row:
                row['age'] = int(row['age'])
            yield row

# CSV处理示例
for row in Loader().load_from_csv('sample.csv'):
    print(row)

# Excel处理示例
for row in Loader().load_from_excel('sample.xlsx'):
    print(row)

进阶：可注入的处理逻辑 #

如果需要更灵活的处理方式，可以允许传入自定义处理器函数，进一步解耦读取与处理逻辑：

class Loader():
    def load_from_csv(self, csv_path: str, processor=None):
        # 默认使用内置处理器，也可传入自定义函数
        processor = processor or self._default_processor
        with open(csv_path, mode='r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            for row in reader:
                yield processor(row)
    
    @staticmethod
    def _default_processor(row):
        return row
    
    # 自定义处理器示例：转换日期格式
    @staticmethod
    def date_formatter(row):
        if 'join_date' in row:
            row['join_date'] = row['join_date'].replace('/', '-')
        return row

# 使用自定义处理器
loader = Loader()
for row in loader.load_from_csv('sample.csv', processor=Loader.date_formatter):
    print(row)

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