From 3430c9424013194858eede82fc25173911c5e986 Mon Sep 17 00:00:00 2001
From: Rami Abdelrazzaq <ramiabdelrazzaq@gmail.com>
Date: Sun, 13 Sep 2026 12:11:43 -0500
Subject: [PATCH] feat: add Python CSV export helper

---
 sqlite_utils/export.py | 36 ++++++++++++++++++++++++++++++++++++
 1 file changed, 36 insertions(+)
 create mode 100644 sqlite_utils/export.py

diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py
new file mode 100644
index 0000000..cf913d4
--- /dev/null
+++ b/sqlite_utils/export.py
@@ -0,0 +1,36 @@
+import csv
+from typing import Any, Iterable, Optional, Sequence, TextIO, Union
+
+
+CsvDialect = Union[str, csv.Dialect]
+
+
+def rows_to_csv_file(
+    cursor: Any,
+    file: TextIO,
+    *,
+    header: bool = True,
+    dialect: CsvDialect = "excel",
+    **writer_kwargs: Any,
+) -> None:
+    """Write the rows from a DB-API cursor to a CSV-compatible text stream.
+
+    ``cursor`` should be a DB-API cursor whose ``description`` attribute
+    contains the result column metadata. Pass ``dialect="excel-tab"`` to
+    produce TSV output, or provide any dialect accepted by ``csv.writer``.
+
+    Additional keyword arguments are forwarded to ``csv.writer``.
+    """
+    writer = csv.writer(file, dialect=dialect, **writer_kwargs)
+    if header:
+        description: Optional[Sequence[Sequence[Any]]] = cursor.description
+        if description is None:
+            raise ValueError("Cursor does not have result columns")
+        writer.writerow([column[0] for column in description])
+    writer.writerows(_rows(cursor))
+
+
+def _rows(cursor: Iterable[Sequence[Any]]) -> Iterable[Sequence[Any]]:
+    """Keep row iteration lazy so large query results are streamed."""
+    for row in cursor:
+        yield row