如何重新打开已关闭的io.BytesIO对象？解决FTP上传报错

我有两段代码：一段用于在内存中生成CSV文件，另一段用于将该文件发送至FTP服务器。

内存生成CSV的代码 #

def to_csv(events: list) -> io.BytesIO():
    if not events:
        return None
    bio = io.BytesIO()
    iow = io.TextIOWrapper(bio)
    writer = csv.DictWriter(iow, fieldnames=events[0].keys())
    writer.writeheader()
    writer.writerows(events)
    iow.flush()
    bio.seek(0)
    return bio

FTP发送代码 #

def send_data(self, bytes: io.BytesIO) -> str:
    filename = f"{time.time()}.csv"
    if not bytes:
        self.__logger.warning("No data to send")
        return None
    try:
        self.__ftp.storbinary(f"STOR {filename}", bytes)
    except ftp.all_errors as e:
        self.__logger.error(
            "FTP fail data send",
            extra={
                "host": self.__cfg.ftp.host,
                "type": type(e).__name__,
                "line": e.__traceback__.tb_lineno,
                "file": __file__,
                "detail": e,
            },
        )
        return None

现在传入send_data的io.BytesIO对象已关闭，调用storbinary时触发报错：

ValueError: I/O operation on closed file.

我尝试用open(bytes, "rb")重新打开该对象，结果报错：

expected str, bytes or os.PathLike object, not _io.BytesIO

完整报错堆栈：

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 171, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 189, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/accounting/main.py", line 29, in process_events
    result = Uploader(logger, cfg).send_data(csv_file)
  File "/opt/airflow/dags/accounting/packages/uploader.py", line 57, in send_data
    self.__ftp.storbinary(f"STOR {filename}", bytes)
  File "/usr/local/lib/python3.7/ftplib.py", line 513, in storbinary
    buf = fp.read(blocksize)
ValueError: I/O operation on closed file.

1. 先明确核心问题 #

TextIOWrapper默认会在关闭时连带关闭底层的BytesIO对象，加上Airflow上下文可能自动处理文件对象，导致传入send_data的BytesIO已处于关闭状态。另外，open()仅用于磁盘文件操作，无法直接打开BytesIO对象。

2. 具体修复方案 #

方案一：避免BytesIO被提前关闭

修改to_csv函数，创建TextIOWrapper时指定closefd=False，防止关闭文本包装器时连带关闭底层的BytesIO：

def to_csv(events: list) -> io.BytesIO():
    if not events:
        return None
    bio = io.BytesIO()
    # 添加closefd=False，隔离TextIOWrapper和BytesIO的关闭操作
    iow = io.TextIOWrapper(bio, closefd=False)
    writer = csv.DictWriter(iow, fieldnames=events[0].keys())
    writer.writeheader()
    writer.writerows(events)
    iow.flush()
    bio.seek(0)
    return bio

方案二：基于已关闭对象的数据重新创建BytesIO

BytesIO即使关闭，仍可通过getvalue()获取存储的字节数据，基于此重新创建可用的BytesIO对象：

def send_data(self, bytes_obj: io.BytesIO) -> str:
    filename = f"{time.time()}.csv"
    if not bytes_obj:
        self.__logger.warning("No data to send")
        return None
    try:
        # 尝试重置指针，若触发关闭错误则重建对象
        try:
            bytes_obj.seek(0)
        except ValueError:
            data = bytes_obj.getvalue()
            bytes_obj = io.BytesIO(data)
        self.__ftp.storbinary(f"STOR {filename}", bytes_obj)
    except ftp.all_errors as e:
        self.__logger.error(
            "FTP fail data send",
            extra={
                "host": self.__cfg.ftp.host,
                "type": type(e).__name__,
                "line": e.__traceback__.tb_lineno,
                "file": __file__,
                "detail": e,
            },
        )
        return None

方案三：提前备份字节数据

在调用send_data前，先提取BytesIO中的数据，后续用新的BytesIO对象传入：

# 生成CSV后备份数据
csv_file = to_csv(events)
if csv_file:
    csv_data = csv_file.getvalue()
    # 用备份数据创建新的BytesIO传入send_data
    result = Uploader(logger, cfg).send_data(io.BytesIO(csv_data))

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