Skip to content

csv

django_spire.contrib.sync.file.writer.csv

CsvWriter

Bases: Writer

Source code in django_spire/contrib/sync/file/writer/csv.py 
12
13
14
15
16
17
18
19
20
21
22
def __init__(
    self,
    delimiter: str = ',',
    encoding: str = 'utf-8',
    field_map: dict[str, str] | None = None,
    fieldnames: list[str] | None = None,
) -> None:
    self._delimiter = delimiter
    self._encoding = encoding
    self._field_map = field_map or {}
    self._fieldnames = fieldnames

write

Source code in django_spire/contrib/sync/file/writer/csv.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def write(self, file_path: str | Path, records: list[dict[str, Any]]) -> None:
    file_path = Path(file_path)
    mapped = [self._map_record(r) for r in records]

    fieldnames = self._fieldnames

    if fieldnames is None and mapped:
        fieldnames = list(mapped[0].keys())

    if not fieldnames:
        file_path.write_text('', encoding=self._encoding)
        return

    with open(file_path, 'w', encoding=self._encoding, newline='') as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=fieldnames,
            delimiter=self._delimiter,
        )

        writer.writeheader()
        writer.writerows(mapped)