Python异步写入CSV文件发生异常时如何删除不完整损坏文件

核心实现思路是先写入临时文件，所有数据处理写入完成后再将临时文件重命名为目标文件，如果中途出现任何异常直接删除临时文件即可，完全避免生成不完整的目标文件，也不需要提前枚举所有可能的异常类型。

完整实现代码 #

from csv import DictWriter
import asyncio
import os
from tempfile import NamedTemporaryFile

# 目标文件路径
TARGET_FILE = "my_file"
# 提前定义CSV表头，避免循环中重复生成
CSV_KEYS = ["你的字段1", "你的字段2", "你的字段3"] # 替换为实际的表头字段

async def write_data():
    tmp_path = None
    try:
        # 生成和目标文件同目录的临时文件，关闭时不自动删除，方便后续重命名
        with NamedTemporaryFile(mode='w', newline='', dir=os.path.dirname(TARGET_FILE), delete=False) as f:
            tmp_path = f.name
            dict_writer = DictWriter(f, CSV_KEYS)
            # 如需写入表头可添加：dict_writer.writeheader()
            async for item in item_feed:
                processed_item = ... # 保留你的原有数据处理逻辑
                dict_writer.writerow(processed_item)
        # 所有数据写入无异常，将临时文件重命名为目标文件
        if os.path.exists(TARGET_FILE):
            os.remove(TARGET_FILE)
        os.rename(tmp_path, TARGET_FILE)
    except Exception:
        # 任意异常触发时删除存在的临时文件
        if tmp_path and os.path.exists(tmp_path):
            os.remove(tmp_path)
        # 可继续抛出异常方便上层逻辑处理
        raise

if __name__ == "__main__":
    asyncio.run(write_data())

方案说明 #

• 无需手动枚举所有异常类型，你提到的OSError、KeyboardInterrupt、ConnectionError、BrokenPipeError等所有异常都会触发临时文件清理逻辑
• 临时文件和目标文件放在同磁盘目录下，重命名操作为原子操作，不会出现重命名中途故障导致文件损坏的问题
• 如果你的需求是追加写入原有目标文件，可在创建临时文件后，先把原目标文件的内容写入临时文件，再追加新数据后执行重命名即可

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