You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何使用csv模块实现可逐行生成CSV内容的生成器(避免全量内存缓冲)

如何使用csv模块实现可逐行生成CSV内容的生成器(避免全量内存缓冲)

嗨,我懂你想要的是什么——一个不用把整个CSV都塞进内存,能逐行输出内容的生成器,还得支持csv模块的各种配置参数对吧?其实完全可以基于你已经熟悉的csv.writer来改造,用一个临时的小缓冲区就能实现,根本不需要一次性生成整个CSV字符串。

下面是完整的实现方案,完美匹配你想要的API,还能兼容csv.writer的所有参数:

import csv
import io
from typing import Generator


def get_csv_lines(rows: list[list[str]], **kwargs) -> Generator[str, None, None]:
    """
    Generates lines for a CSV file from a list of rows, without buffering the entire CSV in memory.

    Args:
        rows (list[list[str]]): A list where each element is a list of strings representing a row
                                in the CSV. All rows must have the same number of items (columns)
                                to ensure a correctly formatted CSV.
        **kwargs: Additional keyword arguments to pass to the csv.writer constructor 
                  (e.g., dialect, delimiter, quotechar, quoting).

    Yields:
        str: A line of the CSV, without the trailing newline character.

    Examples:
        >>> rows = [["Name", "Age", "City"], ["Alice", "30", "New York"], ["Bob", "25", "Los Angeles"]]
        >>> list(get_csv_lines(rows))
        ['Name,Age,City', 'Alice,30,New York', 'Bob,25,Los Angeles']
        
        >>> # 使用自定义分隔符
        >>> list(get_csv_lines(rows, delimiter=';'))
        ['Name;Age;City', 'Alice;30;New York', 'Bob;25;Los Angeles']
    """
    with io.StringIO() as buffer:
        # 创建csv writer实例,支持传入自定义参数
        writer = csv.writer(buffer, **kwargs)
        for row in rows:
            # 将当前行写入临时缓冲区
            writer.writerow(row)
            # 将缓冲区指针移到开头,读取内容
            buffer.seek(0)
            # 去掉writerow自动添加的换行符,生成一行内容
            line = buffer.read().rstrip('\n')
            yield line
            # 清空缓冲区,准备处理下一行
            buffer.truncate(0)

工作原理说明

  • 我们用io.StringIO作为临时小缓冲区,每次只存储当前行的CSV内容,不会积累整个文件
  • 复用csv.writer的成熟逻辑,自动处理CSV的转义规则(比如包含逗号、引号的字段),不用自己手动实现复杂的格式处理
  • 通过**kwargs参数,你可以传入任何csv.writer支持的配置,比如dialect="excel"、delimiter="\t"、quotechar="'"等等

你想要的使用方式完全可以实现:

rows = [["Name", "Age", "City"], ["Alice", "30", "New York"], ["Bob", "25", "Los Angeles"]]

# 逐行输出,不占用大量内存
for line in get_csv_lines(rows):
    print(line)

# 带自定义dialect的用法
for line in get_csv_lines(rows, dialect="excel"):
    print(line)

这种方式既满足了逐行生成的需求,又避免了全量内存缓冲的问题,还完美兼容csv模块的所有特性,非常靠谱。

备注:内容来源于stack exchange,提问作者Nils

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.13 20:05:28