如何将Python同步生成Zip压缩包的代码改为异步实现？

正确的异步代码实现 #

import asyncio
import io
from aiocsv import AsyncWriter
from aiozipstream import AsyncZipStream, ZIP_DEFLATED

async def write_to_csv_async(asset_list, header) -> io.BytesIO:
    buffer = io.BytesIO()
    # 初始化异步ZIP流，启用压缩
    zip_stream = AsyncZipStream(compression=ZIP_DEFLATED)
    # 向ZIP流中添加要创建的CSV文件
    csv_file_stream = zip_stream.add("result.csv")
    
    # 使用aiocsv的异步Writer写入内容
    writer = AsyncWriter(csv_file_stream, newline="", encoding="utf-8")
    await writer.writerow(header)
    for asset in asset_list:
        await writer.writerow(asset)
    # 必须关闭CSV文件流，确保内容被完整写入ZIP
    await csv_file_stream.close()
    
    # 异步读取ZIP流的分块数据，写入BytesIO
    async for chunk in zip_stream:
        buffer.write(chunk)
    buffer.seek(0)
    return buffer

关键注意事项 #

• 禁止同步异步混用：不要用io.TextIOWrapper这类同步IO工具包裹异步流，会直接阻塞事件循环，导致异步逻辑失效。直接依赖aiocsv.AsyncWriter处理编码、换行等CSV格式逻辑。
• 必须等待异步操作完成：所有writerow、close调用都要加await，确保数据被完整写入流，否则会出现文件内容不完整或损坏的情况。
• 大数据优化：如果asset_list数据量极大，可将循环写入改为异步生成器分批输出，降低内存占用。
• 压缩参数选择：ZIP_DEFLATED启用压缩，若追求极致速度可改用ZIP_STORED（无压缩模式）。

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