本文介绍如何利用 Python 的 csv 模块和 io.StringIO 实现一个真正的生成器函数，逐行生成 CSV 字符串，避免将整个 CSV 加载到内存中，同时支持自定义 dialect、delimiter 等 csv.writer 参数。
本文介绍如何利用 python 的 `csv` 模块和 `io.stringio` 实现一个真正的生成器函数，逐行生成 csv 字符串，避免将整个 csv 加载到内存中，同时支持自定义 `dialect`、`delimiter` 等 `csv.writer` 参数。
要实现一个流式、低内存开销的 CSV 行生成器，关键在于：每次只处理一行数据，并立即 yield 其序列化结果，而不是累积全部内容。csv.writer 本身不直接支持“写入单行并返回字符串”，但我们可以借助 io.StringIO 作为可重用的内存缓冲区，配合 seek(0) 和 truncate() 实现高效复用。
以下是完整、健壮且符合类型提示的实现：
import csv
import io
from typing import Generator, List, Any, Optional, Union
def get_csv_lines(
rows: List[List[Any]],
**writer_kwargs
) -> Generator[str, None, None]:
"""
逐行生成 CSV 格式字符串，不缓存全部内容，内存友好。
Args:
rows: 二维列表，每个子列表代表 CSV 中的一行（字段为任意可字符串化对象）。
**writer_kwargs: 传递给 csv.writer 的参数，如 dialect、delimiter、quotechar、quoting 等。
Yields:
str: 每行对应的 CSV 字符串（不含尾部换行符；若需换行，请自行添加 '\n'）。
Example:
>>> rows = [["Name", "Age", "City"], ["Alice", 30, "New York"], ["Bob", 25, "Los Angeles"]]
>>> list(get_csv_lines(rows, quoting=csv.QUOTE_MINIMAL))
['Name,Age,City', 'Alice,30,"New York"', 'Bob,25,"Los Angeles"']
"""
buf = io.StringIO()
writer = csv.writer(buf, **writer_kwargs)
for row in rows:
# 清空缓冲区：回到开头并截断内容
buf.seek(0)
buf.truncate()
# 写入当前行
writer.writerow(row)
# 获取该行生成的字符串（去除末尾自动添加的 \r\n 或 \n）
line = buf.getvalue().rstrip('\r\n')
yield line
✅ 核心技巧说明：
-
buf.seek(0)将文件指针移回缓冲区起始位置； -
buf.truncate()删除从当前位置到末尾的所有内容（即清空缓冲区）； -
writer.writerow(row)将当前行写入缓冲区； -
buf.getvalue().rstrip('\r\n')提取结果并移除csv.writer自动附加的行终止符（因writerow总会写入换行符），确保每yield的是纯净的 CSV 行字符串。
⚠️ 注意事项：
- 此方案不适用于超长单行（如含巨量字段或超大字符串），但对常规行长度完全安全；
- 若需兼容 Windows 换行（
\r\n）或 Unix 风格（\n），建议在消费端统一处理（如print(line)会自动补\n）； -
writer_kwargs支持所有csv.writer参数，例如dialect="excel-tab"、delimiter=";"、quoting=csv.QUOTE_ALL等，灵活适配不同格式需求； - 输入
row中的元素会被自动str()转换，因此支持数字、布尔值、None等（None转为字符串"None"；如需空字符串映射，需预处理）。
? 典型使用场景：
- 流式响应 Web API（如 Flask/FastAPI 返回 CSV 流）；
- 处理大数据集时逐行写入文件或发送至消息队列；
- 单元测试中快速构造 CSV 片段，避免磁盘 I/O。
通过该生成器，你真正获得了“按需计算、即产即用”的 CSV 行流能力——既保持了标准 csv 模块的健壮性与兼容性，又彻底规避了全量内存驻留问题。
