Skip to content

writer

django_spire.contrib.sync.file.writer

__all__ = ['CsvWriter', 'Writer', 'XmlWriter'] module-attribute

Writer

Bases: Protocol

write

Source code in django_spire/contrib/sync/file/writer/base.py 
10
def write(self, file_path: str | Path, records: list[dict[str, Any]]) -> None: ...

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)

XmlWriter

Bases: Writer

Source code in django_spire/contrib/sync/file/writer/xml.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    root_tag: str,
    record_tag: str,
    fields: list[XmlField | XmlListField],
    encoding: str = 'utf-8',
) -> None:
    if not root_tag:
        message = 'root_tag must not be empty'
        raise FileSyncParameterError(message)

    if not record_tag:
        message = 'record_tag must not be empty'
        raise FileSyncParameterError(message)

    if not fields:
        message = 'fields must not be empty'
        raise FileSyncParameterError(message)

    self._root_tag = root_tag
    self._record_tag = record_tag
    self._encoding = encoding
    self._fields: list[XmlField] = []
    self._list_fields: list[XmlListField] = []

    for f in fields:
        if isinstance(f, XmlListField):
            self._list_fields.append(f)
        else:
            self._fields.append(f)

write

Source code in django_spire/contrib/sync/file/writer/xml.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def write(self, file_path: str | Path, records: list[dict[str, Any]]) -> None:
    file_path = Path(file_path)
    root = Element(self._root_tag)

    for record in records:
        element = SubElement(root, self._record_tag)

        for field in self._fields:
            value = record.get(field.key, field.default)
            self._set_field(element, field.path, value)

        for field in self._list_fields:
            values = record.get(field.key, [])
            self._set_list_field(element, field.path, values)

    tree = ElementTree(root)
    indent(tree, space='  ')
    tree.write(str(file_path), encoding=self._encoding, xml_declaration=True)