From 3430c9424013194858eede82fc25173911c5e986 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Sun, 13 Sep 2026 12:11:43 -0500 Subject: [PATCH 1/8] 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 000000000..cf913d4bc --- /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 From 3b7375fcc6d84b717360217afc36089d488e5eb4 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Sun, 13 Sep 2026 12:11:51 -0500 Subject: [PATCH 2/8] test: cover CSV export helper --- tests/test_export.py | 49 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 tests/test_export.py diff --git a/tests/test_export.py b/tests/test_export.py new file mode 100644 index 000000000..bb91022b9 --- /dev/null +++ b/tests/test_export.py @@ -0,0 +1,49 @@ +import io +import sqlite3 + +import pytest + +from sqlite_utils.export import rows_to_csv_file + + +def _cursor(): + db = sqlite3.connect(":memory:") + db.execute("create table creatures (id integer, name text)") + db.executemany( + "insert into creatures (id, name) values (?, ?)", + ((1, "Cleo"), (2, "Cardi, Jr.")), + ) + return db, db.execute("select id, name from creatures order by id") + + +def test_rows_to_csv_file(): + db, cursor = _cursor() + output = io.StringIO(newline="") + rows_to_csv_file(cursor, output, lineterminator="\n") + assert output.getvalue() == 'id,name\n1,Cleo\n2,"Cardi, Jr."\n' + db.close() + + +def test_rows_to_csv_file_without_header(): + db, cursor = _cursor() + output = io.StringIO(newline="") + rows_to_csv_file(cursor, output, header=False, lineterminator="\n") + assert output.getvalue() == '1,Cleo\n2,"Cardi, Jr."\n' + db.close() + + +def test_rows_to_csv_file_tsv(): + db, cursor = _cursor() + output = io.StringIO(newline="") + rows_to_csv_file(cursor, output, dialect="excel-tab", lineterminator="\n") + assert output.getvalue() == "id\tname\n1\tCleo\n2\tCardi, Jr.\n" + db.close() + + +def test_rows_to_csv_file_requires_result_columns_for_header(): + db = sqlite3.connect(":memory:") + cursor = db.execute("create table creatures (id integer)") + output = io.StringIO(newline="") + with pytest.raises(ValueError, match="Cursor does not have result columns"): + rows_to_csv_file(cursor, output) + db.close() From 6f064ce6eeb6f1f4dd1d8a2be009c92577b7f811 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Sun, 13 Sep 2026 13:15:08 -0500 Subject: [PATCH 3/8] fix: validate CSV export cursors before writing --- sqlite_utils/export.py | 17 ++++++----------- tests/test_export.py | 6 ++++-- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py index cf913d4bc..44dbcade8 100644 --- a/sqlite_utils/export.py +++ b/sqlite_utils/export.py @@ -1,5 +1,5 @@ import csv -from typing import Any, Iterable, Optional, Sequence, TextIO, Union +from typing import Any, Sequence, TextIO, Union CsvDialect = Union[str, csv.Dialect] @@ -21,16 +21,11 @@ def rows_to_csv_file( Additional keyword arguments are forwarded to ``csv.writer``. """ + description: Union[Sequence[Sequence[Any]], None] = cursor.description + if description is None: + raise ValueError("Cursor does not have result columns") + 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 + writer.writerows(cursor) diff --git a/tests/test_export.py b/tests/test_export.py index bb91022b9..a72b2bfd4 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -40,10 +40,12 @@ def test_rows_to_csv_file_tsv(): db.close() -def test_rows_to_csv_file_requires_result_columns_for_header(): +@pytest.mark.parametrize("header", (True, False)) +def test_rows_to_csv_file_requires_result_columns(header): db = sqlite3.connect(":memory:") cursor = db.execute("create table creatures (id integer)") output = io.StringIO(newline="") with pytest.raises(ValueError, match="Cursor does not have result columns"): - rows_to_csv_file(cursor, output) + rows_to_csv_file(cursor, output, header=header) + assert output.getvalue() == "" db.close() From 85fe3c3407c26216a6d8c63f50f6a2fedd539326 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Sun, 13 Sep 2026 17:45:06 -0500 Subject: [PATCH 4/8] Fix Black formatting in CSV export helper --- sqlite_utils/export.py | 1 - 1 file changed, 1 deletion(-) diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py index 44dbcade8..9cde7dc98 100644 --- a/sqlite_utils/export.py +++ b/sqlite_utils/export.py @@ -1,7 +1,6 @@ import csv from typing import Any, Sequence, TextIO, Union - CsvDialect = Union[str, csv.Dialect] From fb53549500002e04f9f36d413b5fec406756ccd9 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Mon, 14 Sep 2026 07:04:12 -0500 Subject: [PATCH 5/8] Fix CSV dialect class type support --- sqlite_utils/export.py | 4 ++-- tests/test_export.py | 12 ++++++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py index 9cde7dc98..754fc74ec 100644 --- a/sqlite_utils/export.py +++ b/sqlite_utils/export.py @@ -1,7 +1,7 @@ import csv -from typing import Any, Sequence, TextIO, Union +from typing import Any, Sequence, TextIO, Type, Union -CsvDialect = Union[str, csv.Dialect] +CsvDialect = Union[str, csv.Dialect, Type[csv.Dialect]] def rows_to_csv_file( diff --git a/tests/test_export.py b/tests/test_export.py index a72b2bfd4..b2a03092d 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -1,3 +1,4 @@ +import csv import io import sqlite3 @@ -40,6 +41,17 @@ def test_rows_to_csv_file_tsv(): db.close() +def test_rows_to_csv_file_accepts_dialect_class(): + class PipeDialect(csv.excel): + delimiter = "|" + + db, cursor = _cursor() + output = io.StringIO(newline="") + rows_to_csv_file(cursor, output, dialect=PipeDialect, lineterminator="\n") + assert output.getvalue() == "id|name\n1|Cleo\n2|Cardi, Jr.\n" + db.close() + + @pytest.mark.parametrize("header", (True, False)) def test_rows_to_csv_file_requires_result_columns(header): db = sqlite3.connect(":memory:") From 8af34609f323dc50bc554229cabae7b4ba108701 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Mon, 14 Sep 2026 08:27:46 -0500 Subject: [PATCH 6/8] Handle mapping rows in CSV export helper --- sqlite_utils/export.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py index 754fc74ec..fa8eba971 100644 --- a/sqlite_utils/export.py +++ b/sqlite_utils/export.py @@ -1,4 +1,5 @@ import csv +from collections.abc import Mapping from typing import Any, Sequence, TextIO, Type, Union CsvDialect = Union[str, csv.Dialect, Type[csv.Dialect]] @@ -24,7 +25,16 @@ def rows_to_csv_file( if description is None: raise ValueError("Cursor does not have result columns") + columns = [column[0] for column in description] writer = csv.writer(file, dialect=dialect, **writer_kwargs) if header: - writer.writerow([column[0] for column in description]) - writer.writerows(cursor) + writer.writerow(columns) + + def rows(): + for row in cursor: + if isinstance(row, Mapping): + yield [row[column] for column in columns] + else: + yield row + + writer.writerows(rows()) From 436ac10a1de25ced5e7f4299e7cda2d2d798b4cc Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Mon, 14 Sep 2026 08:28:00 -0500 Subject: [PATCH 7/8] Test CSV export with mapping row factories --- tests/test_export.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/test_export.py b/tests/test_export.py index b2a03092d..5cfbd51f8 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -52,6 +52,28 @@ class PipeDialect(csv.excel): db.close() +def test_rows_to_csv_file_with_mapping_row_factory(): + db = sqlite3.connect(":memory:") + db.execute("create table creatures (id integer, name text)") + db.executemany( + "insert into creatures (id, name) values (?, ?)", + ((1, "Cleo"), (2, "Cardi, Jr.")), + ) + + def dict_factory(cursor, row): + return { + column[0]: value + for column, value in zip(cursor.description, row) + } + + db.row_factory = dict_factory + cursor = db.execute("select id, name from creatures order by id") + output = io.StringIO(newline="") + rows_to_csv_file(cursor, output, lineterminator="\n") + assert output.getvalue() == 'id,name\n1,Cleo\n2,"Cardi, Jr."\n' + db.close() + + @pytest.mark.parametrize("header", (True, False)) def test_rows_to_csv_file_requires_result_columns(header): db = sqlite3.connect(":memory:") From ffb510c56ae922c294837bf88a2c7836a1078854 Mon Sep 17 00:00:00 2001 From: Rami Abdelrazzaq Date: Mon, 14 Sep 2026 12:11:00 -0500 Subject: [PATCH 8/8] Reject ambiguous mapping CSV rows --- sqlite_utils/export.py | 22 +++++++++++++++------- tests/test_export.py | 24 ++++++++++++++++++++---- 2 files changed, 35 insertions(+), 11 deletions(-) diff --git a/sqlite_utils/export.py b/sqlite_utils/export.py index fa8eba971..a5a8123f0 100644 --- a/sqlite_utils/export.py +++ b/sqlite_utils/export.py @@ -1,4 +1,5 @@ import csv +import itertools from collections.abc import Mapping from typing import Any, Sequence, TextIO, Type, Union @@ -26,15 +27,22 @@ def rows_to_csv_file( raise ValueError("Cursor does not have result columns") columns = [column[0] for column in description] + row_iterator = iter(cursor) + sentinel = object() + first_row = next(row_iterator, sentinel) + if isinstance(first_row, Mapping) and len(set(columns)) != len(columns): + raise ValueError("Mapping rows cannot represent duplicate column names") + writer = csv.writer(file, dialect=dialect, **writer_kwargs) if header: writer.writerow(columns) - def rows(): - for row in cursor: - if isinstance(row, Mapping): - yield [row[column] for column in columns] - else: - yield row + def normalize_row(row): + if isinstance(row, Mapping): + return [row[column] for column in columns] + return row - writer.writerows(rows()) + if first_row is not sentinel: + writer.writerows( + normalize_row(row) for row in itertools.chain((first_row,), row_iterator) + ) diff --git a/tests/test_export.py b/tests/test_export.py index 5cfbd51f8..44c08d4ed 100644 --- a/tests/test_export.py +++ b/tests/test_export.py @@ -61,10 +61,7 @@ def test_rows_to_csv_file_with_mapping_row_factory(): ) def dict_factory(cursor, row): - return { - column[0]: value - for column, value in zip(cursor.description, row) - } + return {column[0]: value for column, value in zip(cursor.description, row)} db.row_factory = dict_factory cursor = db.execute("select id, name from creatures order by id") @@ -74,6 +71,25 @@ def dict_factory(cursor, row): db.close() +def test_rows_to_csv_file_rejects_duplicate_mapping_columns(): + db = sqlite3.connect(":memory:") + db.execute("create table creatures (id integer, name text)") + db.execute("insert into creatures (id, name) values (1, 'Cleo')") + + def dict_factory(cursor, row): + return {column[0]: value for column, value in zip(cursor.description, row)} + + db.row_factory = dict_factory + cursor = db.execute("select id as value, name as value from creatures") + output = io.StringIO(newline="") + with pytest.raises( + ValueError, match="Mapping rows cannot represent duplicate column names" + ): + rows_to_csv_file(cursor, output, lineterminator="\n") + assert output.getvalue() == "" + db.close() + + @pytest.mark.parametrize("header", (True, False)) def test_rows_to_csv_file_requires_result_columns(header): db = sqlite3.connect(":memory:")