mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-10-23 17:57:03 +02:00
feat: add Python CSV export helper
This commit is contained in:
parent
8d74ffc932
commit
3430c94240
1 changed files with 36 additions and 0 deletions
36
sqlite_utils/export.py
Normal file
36
sqlite_utils/export.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Add a link
Reference in a new issue