如何使用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
相关产品推荐
相关产品推荐

