Coverage for pycommons/io/csv.py: 100%
298 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-15 00:53 +0000
1"""
2Tools for CSV output and input.
4Our CSV format tools are intended to read and write structured objects from
5and to a comma-separated-values format. This format consists of one header,
6where the column titles are included (separated by a :const:`CSV_SEPARATOR`)
7and one row per data object, with one value per column.
9Different from other CSV processing tools, we want to
111. Permit that data is extracted from / parsed in form of hierarchically
12 structured objects.
132. Columns have fixed types based on the object definition.
143. The data read and written is strictly validated during the process.
154. Data can be processed in form of a stream and is not necessarily all loaded
16 into memory at once.
175. The order of the columns is unimportant.
186. Useless white space is automatically stripped and ignored.
197. Multiple objects may be written per row, maybe even nested objects, and
20 this is signified by "scope" column titles, e.g., something like
21 `"weight.min"`, `"weight.median"`, ..., `"age.min"`, `"age.median"`, ...
228. Comments may be added to the header or footer of the CSV file that describe
23 the contents of the columns.
25The separator is configurable, but by default set to :const:`CSV_SEPARATOR`.
26Comments start with a comment start with :const:`COMMENT_START` by default.
27"""
28from typing import (
29 Any,
30 Callable,
31 Final,
32 Generator,
33 Iterable,
34 Mapping,
35 TypeVar,
36 cast,
37)
39from pycommons.ds.sequences import reiterable, unpack_reiterable
40from pycommons.strings.chars import NEWLINE, WHITESPACE_OR_NEWLINE
41from pycommons.types import check_int_range, type_error
42from pycommons.version import __version__ as pycommons_version
44#: the default CSV separator
45CSV_SEPARATOR: Final[str] = ";"
47#: everything after this character is considered a comment
48COMMENT_START: Final[str] = "#"
50#: the separator to be used between scopes for nested column prefixes
51SCOPE_SEPARATOR: Final[str] = "."
53#: the type variable for data to be written to CSV or to be read from CSV
54T = TypeVar("T")
56# mypy: disable-error-code=valid-type
57#: the type variable for the CSV output setup
58S = TypeVar("S")
61def csv_scope(scope: str | None, key: str | None) -> str:
62 """
63 Combine a scope and a key.
65 :param scope: the scope, or `None`
66 :param key: the key, or `None`
67 :returns: the scope joined with the key
69 >>> csv_scope("a", "b")
70 'a.b'
71 >>> csv_scope("a", None)
72 'a'
73 >>> csv_scope(None, "b")
74 'b'
76 >>> try:
77 ... csv_scope(1, "b")
78 ... except TypeError as te:
79 ... print(str(te))
80 descriptor '__len__' requires a 'str' object but received a 'int'
82 >>> try:
83 ... csv_scope("a", 1)
84 ... except TypeError as te:
85 ... print(str(te))
86 descriptor '__len__' requires a 'str' object but received a 'int'
88 >>> try:
89 ... csv_scope("a ", "b")
90 ... except ValueError as ve:
91 ... print(str(ve))
92 Invalid csv scope 'a '.
94 >>> try:
95 ... csv_scope("", "b")
96 ... except ValueError as ve:
97 ... print(ve)
98 Invalid csv scope ''.
100 >>> try:
101 ... csv_scope("a", " b")
102 ... except ValueError as ve:
103 ... print(str(ve))
104 Invalid csv key ' b'.
106 >>> try:
107 ... csv_scope("a", "")
108 ... except ValueError as ve:
109 ... print(str(ve))
110 Invalid csv key ''.
112 >>> try:
113 ... csv_scope(None, None)
114 ... except ValueError as ve:
115 ... print(str(ve))
116 Csv scope and key cannot both be None.
117 """
118 if (key is not None) and ((str.__len__(key) <= 0) or (
119 str.strip(key) != key)):
120 raise ValueError(f"Invalid csv key {key!r}.")
121 if scope is None:
122 if key is None:
123 raise ValueError("Csv scope and key cannot both be None.")
124 return key
125 if (str.__len__(scope) <= 0) or (str.strip(scope) != scope):
126 raise ValueError(f"Invalid csv scope {scope!r}.")
127 if key is None:
128 return scope
129 return f"{scope}{SCOPE_SEPARATOR}{key}"
132def csv_read(rows: Iterable[str],
133 setup: Callable[[dict[str, int]], S],
134 parse_row: Callable[[S, list[str]], T],
135 separator: str = CSV_SEPARATOR,
136 comment_start: str | None = COMMENT_START) \
137 -> Generator[T, None, None]:
138 r"""
139 Read (parse) a sequence of strings as CSV data.
141 All lines str :meth:`~str.split` based on the `separator` string and each
142 of the resulting strings is stripped via :meth:`~str.strip`.
143 The first non-empty line of the data is interpreted as header line.
145 This header is passed to the `setup` function in form of a :class:`dict`
146 that maps column titles to column indices. This function then returns an
147 object of setup data. To each of the rows of CSV data, the function
148 `parse_row` is applied. This function receives the object returned by
149 `setup` as first argument and the row as list of strings as second
150 argument. Each line is therefore :meth:`~str.split` (by the CSV separator)
151 and its component :meth:`~str.strip`-ped.
152 It is permitted that a line in the CSV file contains fewer columns than
153 declared in the header. In this case, the missing columns are set to empty
154 strings. Lines that are entirely empty are skipped.
156 If `comment_start` is not none, then all text in a line starting at the
157 first occurence of `comment_start` is discarted before the line is
158 processed.
160 If you want to read more complex CSV structures, then using the class
161 :class:`CsvReader` and its class method :meth:`CsvReader.read` are a more
162 convenient approach. They are wrappers around :func:`csv_read`.
164 :param rows: the rows of text
165 :param setup: a function which creates an object holding the necessary
166 information for row parsing
167 :param parse_row: the unction parsing the rows
168 :param separator: the string used to separate columns
169 :param comment_start: the string starting comments
170 :returns: an :class:`Generator` with the parsed data rows
171 :raises TypeError: if any of the parameters has the wrong type
172 :raises ValueError: if the separator or comment start character are
173 incompatible or if the data has some internal error
175 >>> def _setup(colidx: dict[str, int]) -> dict[str, int]:
176 ... return colidx
178 >>> def _parse_row(colidx: dict[str, int], row: list[str]) -> dict:
179 ... return {x: row[y] for x, y in colidx.items()}
181 >>> text = ["a;b;c;d", "# test", " 1; 2;3;4", " 5 ;6 ", ";8;;9",
182 ... "", "10", "# 11;12"]
184 >>> for p in csv_read(text, _setup, _parse_row):
185 ... print(p)
186 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
187 {'a': '5', 'b': '6', 'c': '', 'd': ''}
188 {'a': '', 'b': '8', 'c': '', 'd': '9'}
189 {'a': '10', 'b': '', 'c': '', 'd': ''}
191 >>> for p in csv_read((t.replace(";", ",") for t in text), _setup,
192 ... _parse_row, ","):
193 ... print(p)
194 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
195 {'a': '5', 'b': '6', 'c': '', 'd': ''}
196 {'a': '', 'b': '8', 'c': '', 'd': '9'}
197 {'a': '10', 'b': '', 'c': '', 'd': ''}
199 >>> for p in csv_read((t.replace(";", "\t") for t in text), _setup,
200 ... _parse_row, "\t"):
201 ... print(p)
202 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
203 {'a': '5', 'b': '6', 'c': '', 'd': ''}
204 {'a': '', 'b': '8', 'c': '', 'd': '9'}
205 {'a': '10', 'b': '', 'c': '', 'd': ''}
207 >>> for p in csv_read(text, _setup, _parse_row, comment_start=None):
208 ... print(p)
209 {'a': '# test', 'b': '', 'c': '', 'd': ''}
210 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
211 {'a': '5', 'b': '6', 'c': '', 'd': ''}
212 {'a': '', 'b': '8', 'c': '', 'd': '9'}
213 {'a': '10', 'b': '', 'c': '', 'd': ''}
214 {'a': '# 11', 'b': '12', 'c': '', 'd': ''}
216 >>> text = ["a;b;c;d", "# test", " 1; 2;3;4", " 5 ;6 ", "5;6", ";8;;9",
217 ... "", "10", "# 11;12"]
218 >>> for p in csv_read(text, _setup, _parse_row):
219 ... print(p)
220 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
221 {'a': '5', 'b': '6', 'c': '', 'd': ''}
222 {'a': '5', 'b': '6', 'c': '', 'd': ''}
223 {'a': '', 'b': '8', 'c': '', 'd': '9'}
224 {'a': '10', 'b': '', 'c': '', 'd': ''}
226 >>> try:
227 ... list(csv_read(None, _setup, _parse_row))
228 ... except TypeError as te:
229 ... print(te)
230 rows should be an instance of typing.Iterable but is None.
232 >>> try:
233 ... list(csv_read(1, _setup, _parse_row))
234 ... except TypeError as te:
235 ... print(te)
236 rows should be an instance of typing.Iterable but is int, namely 1.
238 >>> try:
239 ... list(csv_read(text, None, _parse_row))
240 ... except TypeError as te:
241 ... print(te)
242 setup should be a callable but is None.
244 >>> try:
245 ... list(csv_read(text, 1, _parse_row))
246 ... except TypeError as te:
247 ... print(te)
248 setup should be a callable but is int, namely 1.
250 >>> try:
251 ... list(csv_read(text, _setup, None))
252 ... except TypeError as te:
253 ... print(te)
254 parse_row should be a callable but is None.
256 >>> try:
257 ... list(csv_read(text, _setup, 1))
258 ... except TypeError as te:
259 ... print(te)
260 parse_row should be a callable but is int, namely 1.
262 >>> try:
263 ... list(csv_read(text, _setup, _parse_row, None))
264 ... except TypeError as te:
265 ... print(te)
266 descriptor '__len__' requires a 'str' object but received a 'NoneType'
268 >>> try:
269 ... list(csv_read(text, _setup, _parse_row, 1))
270 ... except TypeError as te:
271 ... print(te)
272 descriptor '__len__' requires a 'str' object but received a 'int'
274 >>> try:
275 ... list(csv_read(text, _setup, _parse_row, ""))
276 ... except ValueError as ve:
277 ... print(ve)
278 Invalid separator ''.
280 >>> try:
281 ... list(csv_read(text, _setup, _parse_row, "-", 1))
282 ... except TypeError as te:
283 ... print(te)
284 descriptor '__len__' requires a 'str' object but received a 'int'
286 >>> try:
287 ... list(csv_read(text, _setup, _parse_row, "-", ""))
288 ... except ValueError as ve:
289 ... print(ve)
290 Invalid comment start: ''.
292 >>> try:
293 ... list(csv_read(text, _setup, _parse_row, "-", " "))
294 ... except ValueError as ve:
295 ... print(ve)
296 Invalid comment start: ' '.
298 >>> try:
299 ... list(csv_read(text, _setup, _parse_row, ";", ";"))
300 ... except ValueError as ve:
301 ... print(ve)
302 Invalid comment start: ';'.
304 >>> text2 = ["a;b;a;d", "# test", " 1; 2;3;4", " 5 ;6 ", ";8;;9"]
305 >>> try:
306 ... list(csv_read(text2, _setup, _parse_row))
307 ... except ValueError as ve:
308 ... print(ve)
309 Invalid column headers: ['a', 'b', 'a', 'd'].
311 >>> text2 = ["a;b;c;d", "# test", " 1; 2;3;4", "1;2;3;4;5;6;7", ";8;;9"]
312 >>> try:
313 ... list(csv_read(text2, _setup, _parse_row))
314 ... except ValueError as ve:
315 ... print(ve)
316 Invalid row '1;2;3;4;5;6;7' contains 7 columns, but should have at most 4.
317 """
318 if not isinstance(rows, Iterable):
319 raise type_error(rows, "rows", Iterable)
320 if not callable(setup):
321 raise type_error(setup, "setup", call=True)
322 if not callable(parse_row):
323 raise type_error(parse_row, "parse_row", call=True)
324 if str.__len__(separator) <= 0:
325 raise ValueError(f"Invalid separator {separator!r}.")
326 if (comment_start is not None) and (
327 (str.__len__(comment_start) <= 0) or (
328 str.strip(comment_start) != comment_start) or (
329 comment_start in separator)):
330 raise ValueError(f"Invalid comment start: {comment_start!r}.")
332 col_count: int = -1
334 # cannot strip spaces that are part of the separator
335 strip: Final[Callable[[str], str]] = str.strip
336 stripper: Final[Callable[[str], str]] = strip if ( # type: ignore
337 strip(separator) == separator) else str.rstrip # type: ignore
338 find: Final[Callable[[str, str], int]] = str.find # type: ignore
339 split: Final[Callable[[str, str], list[str]]] = str.split # type: ignore
340 listlen: Final[Callable[[list], int]] = list.__len__ # type: ignore
341 strlen: Final[Callable[[str], int]] = str.__len__ # type: ignore
342 info: S | None = None # the column definition info generated by setup
343 exts: dict[int, list[str]] = {} # the list of extensions
345 for orig_line in rows: # iterate over all the rows
346 line: str = orig_line
347 if comment_start is not None: # delete comment part, if any
348 deli = find(line, comment_start)
349 if deli >= 0:
350 line = line[:deli]
351 line = stripper(line)
352 if strlen(line) <= 0:
353 continue # nothing to do here
355 cols: list[str] = split(line, separator) # split into columns
356 for i, v in enumerate(cols): # string whitespace off columns
357 cols[i] = strip(v)
359 if info is None: # need to load column definition
360 col_count = listlen(cols)
361 colmap: dict[str, int] = {s: i for i, s in enumerate(cols)}
362 if any(strlen(s) <= 0 for s in cols) or (
363 dict.__len__(colmap) != col_count) or (col_count <= 0):
364 raise ValueError(f"Invalid column headers: {cols!r}.")
365 info = setup(colmap) # obtain the column setup object
366 del colmap # column map no longer needed
367 continue # proceed with next line
369 count: int = listlen(cols) # get number of columns
370 if count > col_count: # too many columns, throw error
371 raise ValueError(
372 f"Invalid row {orig_line!r} contains {count} columns, but "
373 f"should have at most {col_count}.")
374 if count < col_count: # do we need to add dummy columns?
375 add: int = col_count - count # number of needed columns
376 if add not in exts: # check if in cache
377 exts[add] = [""] * add # add to cache
378 cols.extend(exts[add])
379 yield parse_row(info, cols)
382def pycommons_footer_bottom_comments(
383 _: Any, additional: str | None = None) -> Iterable[str]:
384 """
385 Print standard footer bottom comments for `pycommons`.
387 :param _: ignored
388 :param additional: an optional line of additional comments
389 :returns: an :class:`Iterable` of standard pycommons comments
391 >>> for p in pycommons_footer_bottom_comments(""):
392 ... print(p[:70])
393 This CSV output has been created using the versatile CSV API of pycomm
394 You can find pycommons at https://thomasweise.github.io/pycommons.
396 >>> for p in pycommons_footer_bottom_comments("", "Statistics are cool."):
397 ... print(p[:70])
398 This CSV output has been created using the versatile CSV API of pycomm
399 Statistics are cool.
400 You can find pycommons at https://thomasweise.github.io/pycommons.
401 """
402 yield ("This CSV output has been created using the versatile CSV API of "
403 f"pycommons.io.csv, version {pycommons_version}.")
404 if (additional is not None) and (str.__len__(additional) > 0):
405 yield additional
406 yield "You can find pycommons at https://thomasweise.github.io/pycommons."
409def __print_comments(comments: Iterable[str] | None,
410 comment_start: str, comment_type: str,
411 empty_first_row: bool) -> Generator[str, None, None]:
412 r"""
413 Produce the comments after formatting and checking them.
415 :param comments: the comment source
416 :param comment_start: the comment start string
417 :param comment_type: the comment type
418 :param empty_first_row: should we put an empty first row?
419 :returns: the generator of the comment strings
420 :raises TypeError: if an argument is of the wrong type
421 :raises ValueError: if comments cannot be placed or contain newlines
423 >>> col = ["", "First comment.", "Second comment.", "", "",
424 ... " Third comment. "]
425 >>> for p in __print_comments(col, "#", "header", False):
426 ... print(p)
427 # First comment.
428 # Second comment.
429 #
430 # Third comment.
432 >>> col.clear()
433 >>> list(__print_comments(col, "#", "header", True))
434 []
436 >>> col = ["", "First comment.", "Second comment.", "", "",
437 ... " Third comment. "]
438 >>> for p in __print_comments(col, "#", "header", True):
439 ... print(p)
440 #
441 # First comment.
442 # Second comment.
443 #
444 # Third comment.
446 >>> col = ["First comment.", "Second comment.", "", "",
447 ... " Third comment. "]
448 >>> for p in __print_comments(col, "#", "header", True):
449 ... print(p)
450 #
451 # First comment.
452 # Second comment.
453 #
454 # Third comment.
456 >>> col = ["", "", "First comment.", "Second comment.", "", "",
457 ... " Third comment. "]
458 >>> for p in __print_comments(col, "#", "header", True):
459 ... print(p)
460 #
461 # First comment.
462 # Second comment.
463 #
464 # Third comment.
466 >>> list(__print_comments([], "#", "header", False))
467 []
468 >>> list(__print_comments([""], "#", "header", False))
469 []
470 >>> list(__print_comments(["", ""], "#", "header", False))
471 []
472 >>> list(__print_comments([], "#", "header", True))
473 []
474 >>> list(__print_comments([""], "#", "header", True))
475 []
476 >>> list(__print_comments(["", ""], "#", "header", True))
477 []
479 >>> list(__print_comments(None, "#", "header", True))
480 []
482 >>> try:
483 ... list(__print_comments(1, "#", "header", True))
484 ... except TypeError as te:
485 ... print(te)
486 comments should be an instance of typing.Iterable but is int, namely 1.
488 >>> try:
489 ... list(__print_comments(["", 1, "Second comment."], "x", "header",
490 ... False))
491 ... except TypeError as te:
492 ... print(te)
493 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
495 >>> try:
496 ... list(__print_comments(["", None, "Second."], "x", "header",
497 ... False))
498 ... except TypeError as te:
499 ... print(te)
500 descriptor 'strip' for 'str' objects doesn't apply to a 'NoneType' object
502 >>> try:
503 ... list(__print_comments(["Hello", "x\ny", "z"], "#", "header",
504 ... False))
505 ... except ValueError as ve:
506 ... print(ve)
507 A header comment must not contain a newline character, but 'x\ny' does.
508 """
509 if comments is None:
510 return
511 if not isinstance(comments, Iterable):
512 raise type_error(comments, "comments", Iterable)
513 not_first = False
514 for cmt in comments:
515 xcmt = str.strip(cmt) # strip and typecheck
516 if str.__len__(xcmt) <= 0:
517 if not_first:
518 yield comment_start
519 empty_first_row = not_first = False
520 continue
521 if any(map(xcmt.__contains__, NEWLINE)):
522 raise ValueError(f"A {comment_type} comment must not contain "
523 f"a newline character, but {cmt!r} does.")
524 not_first = True
525 if empty_first_row:
526 yield comment_start
527 empty_first_row = False
528 yield f"{comment_start} {xcmt}"
531def __default_row(s: Iterable[str], t: Any) -> Iterable[str]:
532 """
533 Generate row data in the default way.
535 :param s: the setup object: an :class:`Iterable` of string
536 :param t: the row object
537 :returns: an :class:`Iterable` of string
539 >>> list(__default_row(("a", "b"), ("1", "2")))
540 ['1', '2']
542 >>> list(__default_row(("a", "b"), {"b": 45, "c": 44, "a": 6}))
543 ['6', '45']
544 """
545 if isinstance(t, Mapping):
546 return (str(t[ss]) if ss in t else "" for ss in s)
547 return map(str, cast("Iterable[Any]", t))
550def csv_write(
551 data: Iterable[T],
552 column_titles: Iterable[str] | Callable[[S], Iterable[str]] =
553 lambda t: cast("Iterable[str]", t),
554 get_row: Callable[[S, T], Iterable[str]] =
555 cast("Callable[[S, T], Iterable[str]]", __default_row),
556 setup: Callable[[Iterable[T]], S] = lambda t: cast("S", t),
557 separator: str = CSV_SEPARATOR,
558 comment_start: str | None = COMMENT_START,
559 header_comments:
560 Iterable[str] | Callable[[S], Iterable[str] | None] | None = None,
561 footer_comments:
562 Iterable[str] | Callable[[S], Iterable[str] | None] | None = None,
563 footer_bottom_comments: Iterable[str] | Callable[[
564 S], Iterable[str] | None] | None =
565 pycommons_footer_bottom_comments) -> Generator[str, None, None]:
566 r"""
567 Produce a sequence of CSV formatted text.
569 The data is provided in form of a :class:`Iterable`. In a first step, the
570 function `setup` is invoked and applied to the `data` :class:`Iterable`.
571 It can return an object that sort of stores the structure of the data,
572 e.g., which columns should be generated and how they should be formatted.
574 `column_titles` can either be an :class:`Iterable` with the column titles
575 or a :class:`Callable`. In the latter case, the object generated by `setup`
576 is passed to `column_titles`, which should generate the column titles.
577 These titles are :meth:`~str.strip`-ped and concatenated to use the column
578 `separator` string and the resulting header string is passed to `consumer`.
580 Then, for each element `e` in the `data` :class:`Iterable`, the function
581 `get_row` is invoked. This function receives the setup information object
582 (previously returned by `setup`). It should generate one string per
583 column. These strings are then each :meth:`~str.strip`-ped and
584 concatenated using the column `separator` string. All trailing `separator`
585 are removed, but if all strings are empty, at least a single `separator`
586 is retained. The resulting string (per row) is again passed to `consumer`.
588 Additionally, `header_comments` and `footer_comments` can be `None`, to
589 not include any such comments, an :class:`Iterable` of comments, or
590 functions to generate row comments as :class:`str`. These are then
591 prepended or appends as comment rows before or after all of the
592 above, respectively. In that case, `comment_start` is prepended to each
593 line. If `comment_start is None`, then these comments are not printed.
594 `footer_bottom_comments` provides means to print additional comments
595 after the footer comments `comment_start is not None`.
597 If you create nested CSV formats, i.e., such where the `setup` function
598 invokes the `setup` function of other data, and the data that you receive
599 could come from a :class:`~typing.Generator` (or some other one-shot
600 :class:`~typing.Iterator`), then you need to make sure to solidify the
601 iterable data with :func:`~pycommons.ds.sequences.reiterable`. The
602 structure of our CSV output is that `setup` is first invoked and then
603 `get_row`. If `setup` already consumes the data away, then `get_row` may
604 print nothing. Alternatively, if you apply multiple `setup` routines to
605 the same data that extract different information, then the first `setup`
606 run may consume all the data, leaving nothing for the second one.
608 If you want to write more complex CSV structures, then implementing the
609 class :class:`CsvWriter` and using its class method
610 :meth:`CsvWriter.write` may be a more convenient solution.
611 They are wrappers around :func:`csv_write`.
613 :param data: the iterable of data to be written
614 :param column_titles: get the column titles
615 :param get_row: transform a row of data into a list of strings
616 :param setup: the setup function that computes how the data should be
617 represented
618 :param separator: the string used to separate columns
619 :param comment_start: the string starting comments
620 :param header_comments: get the comments to be placed above the CSV
621 header row -- only invoked if `comment_start is not None`.
622 :param footer_comments: get the comments to be placed after the last
623 row -- only invoked if `comment_start is not None`.
624 :param footer_bottom_comments: get the footer bottom comments, i.e.,
625 comments to be printed after all other footers. These commonts may
626 include something like the version information of the software used.
627 This function is only invoked if `comment_start is not None`.
628 :returns: a :class:`Generator` with the rows of CSV text
629 :raises TypeError: if any of the parameters has the wrong type
630 :raises ValueError: if the separator or comment start character are
631 incompatible or if the data has some internal error
633 >>> dd = [{"a": 1, "c": 2}, {"b": 6, "c": 8},
634 ... {"a": 4, "d": 12, "b": 3}, {}]
636 >>> def __setup(datarows) -> list[str]:
637 ... return sorted({dkey for datarow in datarows for dkey in datarow})
639 >>> def __get_row(keyd: list[str], row: dict[str, int]) -> Iterable[str]:
640 ... return map(str, (row.get(key, "") for key in keyd))
642 >>> def __get_header_cmt(keyd: list[str]) -> list[str]:
643 ... return ["This is a header comment.", " We have two of it. "]
645 >>> def __get_footer_cmt(keyd: list[str]) -> list[str]:
646 ... return [" This is a footer comment."]
648 >>> for p in csv_write(dd, lambda x: x, __default_row, __setup,
649 ... ";", "#", __get_header_cmt, __get_footer_cmt,
650 ... lambda _: ()):
651 ... print(p)
652 # This is a header comment.
653 # We have two of it.
654 a;b;c;d
655 1;;2
656 ;6;8
657 4;3;;12
658 ;
659 # This is a footer comment.
661 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
662 ... ";", "#", __get_header_cmt, __get_footer_cmt):
663 ... print(p[:70])
664 # This is a header comment.
665 # We have two of it.
666 a;b;c;d
667 1;;2
668 ;6;8
669 4;3;;12
670 ;
671 # This is a footer comment.
672 #
673 # This CSV output has been created using the versatile CSV API of pyco
674 # You can find pycommons at https://thomasweise.github.io/pycommons.
676 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
677 ... ",", "@@", __get_header_cmt, __get_footer_cmt,
678 ... lambda _: ()):
679 ... print(p)
680 @@ This is a header comment.
681 @@ We have two of it.
682 a,b,c,d
683 1,,2
684 ,6,8
685 4,3,,12
686 ,
687 @@ This is a footer comment.
689 >>> try:
690 ... list(csv_write(None, lambda x: x, __get_row, __setup,
691 ... ";", "#", __get_header_cmt, __get_footer_cmt))
692 ... except TypeError as te:
693 ... print(str(te)[:60])
694 source should be an instance of any in {typing.Iterable, typ
696 >>> try:
697 ... list(csv_write(1, lambda x: x, __get_row, __setup,
698 ... ";", "#", __get_header_cmt, __get_footer_cmt))
699 ... except TypeError as te:
700 ... print(str(te)[:60])
701 source should be an instance of any in {typing.Iterable, typ
703 >>> try:
704 ... list(csv_write(dd, None, __get_row, __setup,
705 ... ";", "#", __get_header_cmt, __get_footer_cmt))
706 ... except TypeError as te:
707 ... print(str(te)[:70])
708 column_titles should be an instance of typing.Iterable or a callable b
710 >>> try:
711 ... list(csv_write(dd, 1, __get_row, __setup,
712 ... ";", "#", __get_header_cmt, __get_footer_cmt))
713 ... except TypeError as te:
714 ... print(str(te)[:70])
715 column_titles should be an instance of typing.Iterable or a callable b
717 >>> try:
718 ... list(csv_write(dd, lambda x: x, None, __setup,
719 ... ";", "#", __get_header_cmt, __get_footer_cmt))
720 ... except TypeError as te:
721 ... print(te)
722 get_row should be a callable but is None.
724 >>> try:
725 ... list(csv_write(dd, lambda x: x, 1, __setup,
726 ... ";", "#", __get_header_cmt, __get_footer_cmt))
727 ... except TypeError as te:
728 ... print(te)
729 get_row should be a callable but is int, namely 1.
731 >>> try:
732 ... list(csv_write(dd, lambda x: x, __get_row, None,
733 ... ";", "#", __get_header_cmt, __get_footer_cmt))
734 ... except TypeError as te:
735 ... print(te)
736 setup should be a callable but is None.
738 >>> try:
739 ... list(csv_write(dd, lambda x: x, __get_row, 1,
740 ... ";", "#", __get_header_cmt, __get_footer_cmt))
741 ... except TypeError as te:
742 ... print(te)
743 setup should be a callable but is int, namely 1.
745 >>> try:
746 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
747 ... None, "#", __get_header_cmt, __get_footer_cmt))
748 ... except TypeError as te:
749 ... print(te)
750 descriptor '__len__' requires a 'str' object but received a 'NoneType'
752 >>> try:
753 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
754 ... 1, "#", __get_header_cmt, __get_footer_cmt))
755 ... except TypeError as te:
756 ... print(te)
757 descriptor '__len__' requires a 'str' object but received a 'int'
759 >>> try:
760 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
761 ... ";", 1, __get_header_cmt, __get_footer_cmt))
762 ... except TypeError as te:
763 ... print(te)
764 descriptor '__len__' requires a 'str' object but received a 'int'
766 >>> try:
767 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
768 ... ";", "#", 1, __get_footer_cmt))
769 ... except TypeError as te:
770 ... print(str(te)[:70])
771 header_comments should be an instance of typing.Iterable or a callable
773 >>> try:
774 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
775 ... ";", "", __get_header_cmt, __get_footer_cmt))
776 ... except ValueError as ve:
777 ... print(ve)
778 Invalid comment start: ''.
780 >>> try:
781 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
782 ... ";", " ", __get_header_cmt, __get_footer_cmt))
783 ... except ValueError as ve:
784 ... print(ve)
785 Invalid comment start: ' '.
787 >>> try:
788 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
789 ... ";", "# ", __get_header_cmt, __get_footer_cmt))
790 ... except ValueError as ve:
791 ... print(ve)
792 Invalid comment start: '# '.
794 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup, ";",
795 ... None, None):
796 ... print(p)
797 a;b;c;d
798 1;;2
799 ;6;8
800 4;3;;12
801 ;
803 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
804 ... ";", None, __get_header_cmt):
805 ... print(p)
806 a;b;c;d
807 1;;2
808 ;6;8
809 4;3;;12
810 ;
812 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
813 ... ";", None, footer_comments=__get_footer_cmt,
814 ... footer_bottom_comments= None):
815 ... print(p)
816 a;b;c;d
817 1;;2
818 ;6;8
819 4;3;;12
820 ;
822 >>> try:
823 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
824 ... ";", "#", __get_header_cmt, 1))
825 ... except TypeError as te:
826 ... print(str(te)[:70])
827 footer_comments should be an instance of typing.Iterable or a callable
829 >>> def __err_cmt_1(keyd: list[str]) -> Iterable[str]:
830 ... return ("This is\n a comment with error.", )
832 >>> try:
833 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
834 ... ";", "#", __err_cmt_1))
835 ... except ValueError as ve:
836 ... print(str(ve)[:58])
837 A header comment must not contain a newline character, but
839 >>> try:
840 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
841 ... ";", "#", footer_comments=__err_cmt_1,
842 ... footer_bottom_comments=None))
843 ... except ValueError as ve:
844 ... print(str(ve)[:58])
845 A footer comment must not contain a newline character, but
847 >>> def __empty_cmt(keyd: list[str]) -> Iterable[str]:
848 ... return (" ", )
850 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
851 ... ";", "#", __empty_cmt, __empty_cmt, __empty_cmt):
852 ... print(p)
853 a;b;c;d
854 1;;2
855 ;6;8
856 4;3;;12
857 ;
859 >>> for p in csv_write(dd, lambda x: x, __get_row, __setup,
860 ... ";", "#", footer_comments=__empty_cmt,
861 ... footer_bottom_comments=lambda _: ()):
862 ... print(p)
863 a;b;c;d
864 1;;2
865 ;6;8
866 4;3;;12
867 ;
869 >>> def __error_column_titles_1(keyd: list[str]) -> Iterable[str]:
870 ... return ()
872 >>> try:
873 ... list(csv_write(dd, __error_column_titles_1, __get_row,
874 ... __setup, ";", "#"))
875 ... except ValueError as ve:
876 ... print(ve)
877 Cannot have zero columns.
879 >>> dde = dd.copy()
880 >>> dde.append(None)
881 >>> try:
882 ... list(csv_write(dde, lambda x: x, __get_row,
883 ... lambda _: ["a", "b", "c", "d"],
884 ... ";", "#", footer_comments=__empty_cmt,
885 ... footer_bottom_comments=lambda _: ()))
886 ... except TypeError as te:
887 ... print(te)
888 data element should be an instance of object but is None.
890 >>> def __error_column_titles_2(keyd: list[str]) -> Iterable[str]:
891 ... return (" ", )
893 >>> try:
894 ... list(csv_write(dd, __error_column_titles_2, __get_row, __setup,
895 ... ";", "#"))
896 ... except ValueError as ve:
897 ... print(str(ve)[:50])
898 Invalid column title ' ', must neither be empty no
900 >>> def __error_column_titles_3(keyd: list[str]) -> Iterable[str]:
901 ... return ("bla\nblugg", )
903 >>> try:
904 ... list(csv_write(dd, __error_column_titles_3, __get_row, __setup,
905 ... ";", "#"))
906 ... except ValueError as ve:
907 ... print(str(ve)[:50])
908 Invalid column title 'bla\nblugg', must neither be
910 >>> def __error_column_titles_4(keyd: list[str]) -> Iterable[str]:
911 ... return (None, )
913 >>> try:
914 ... list(csv_write(dd, __error_column_titles_4, __get_row, __setup,
915 ... ";", "#"))
916 ... except TypeError as te:
917 ... print(te)
918 descriptor 'strip' for 'str' objects doesn't apply to a 'NoneType' object
920 >>> def __error_column_titles_5(keyd: list[str]) -> Iterable[str]:
921 ... return (1, )
923 >>> try:
924 ... list(csv_write(dd, __error_column_titles_5, __get_row, __setup,
925 ... ";", "#"))
926 ... except TypeError as te:
927 ... print(te)
928 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
930 >>> def __error_column_titles_6(keyd: list[str]) -> Iterable[str]:
931 ... return ("a", "b", "c", "a")
933 >>> try:
934 ... list(csv_write(dd, __error_column_titles_6, __get_row, __setup,
935 ... ";", "#"))
936 ... except ValueError as ve:
937 ... print(ve)
938 Cannot have duplicated columns: ['a', 'b', 'c', 'a'].
940 >>> def __error_column_titles_7(keyd: list[str]) -> Iterable[str]:
941 ... return ("a", "b", "c;4")
943 >>> try:
944 ... list(csv_write(dd, __error_column_titles_7, __get_row, __setup,
945 ... ";", "#"))
946 ... except ValueError as ve:
947 ... print(str(ve)[:49])
948 Invalid column title 'c;4', must neither be empty
950 >>> def __error_column_titles_8(keyd: list[str]) -> Iterable[str]:
951 ... return ("a", "b#x", "c")
953 >>> try:
954 ... list(csv_write(dd, __error_column_titles_8, __get_row, __setup,
955 ... ";", "#"))
956 ... except ValueError as ve:
957 ... print(str(ve)[:49])
958 Invalid column title 'b#x', must neither be empty
960 >>> def __error_row_1(keyd: list[str], row: dict[str, int]):
961 ... return ("bla", None, "blubb")
963 >>> try:
964 ... list(csv_write(dd, lambda x: x, __error_row_1,
965 ... __setup, ";", "#",
966 ... footer_bottom_comments=lambda _, __: None))
967 ... except TypeError as te:
968 ... print(te)
969 descriptor 'strip' for 'str' objects doesn't apply to a 'NoneType' object
971 >>> def __error_row_2(keyd: list[str], row: dict[str, int]):
972 ... return ("bla", 2.3, "blubb")
974 >>> try:
975 ... list(csv_write(dd, lambda x: x, __error_row_2,
976 ... __setup, ";", "#",
977 ... footer_bottom_comments=lambda _: None))
978 ... except TypeError as te:
979 ... print(te)
980 descriptor 'strip' for 'str' objects doesn't apply to a 'float' object
982 >>> def __error_row_3(keyd: list[str], row: dict[str, int]):
983 ... return ("bla", "x\ny", "blubb")
985 >>> try:
986 ... list(csv_write(dd, lambda x: x, __error_row_3,
987 ... __setup, ";", "#",
988 ... footer_bottom_comments=lambda _: None))
989 ... except ValueError as ve:
990 ... print(str(ve)[:50])
991 Invalid column value 'x\ny', cannot contain any of
993 >>> def __error_row_4(keyd: list[str], row: dict[str, int]):
994 ... return ("bla", "x#", "blubb")
996 >>> try:
997 ... list(csv_write(dd, lambda x: x, __error_row_4,
998 ... __setup, ";", "#",
999 ... footer_bottom_comments=lambda _: None))
1000 ... except ValueError as ve:
1001 ... print(str(ve)[:50])
1002 Invalid column value 'x#', cannot contain any of [
1004 >>> def __error_row_5(keyd: list[str], row: dict[str, int]):
1005 ... return ("bla", "x;#", "blubb")
1007 >>> try:
1008 ... list(csv_write(dd, lambda x: x, __error_row_5,
1009 ... __setup, ";", "#"))
1010 ... except ValueError as ve:
1011 ... print(str(ve)[:49])
1012 Invalid column value 'x;#', cannot contain any of
1014 >>> def __error_column_titles_9(keyd: list[str]) -> Iterable[str]:
1015 ... return ("a", )
1017 >>> def __error_row_6(keyd: list[str], row: dict[str, int]):
1018 ... return ("", )
1020 >>> try:
1021 ... list(csv_write(dd, __error_column_titles_9, __error_row_6,
1022 ... __setup, ";", "#"))
1023 ... except ValueError as ve:
1024 ... print(ve)
1025 Cannot have empty row in a single-column format, but got [''].
1027 >>> def __error_row_7(keyd: list[str], row: dict[str, int]):
1028 ... return ("x", "y")
1030 >>> try:
1031 ... list(csv_write(dd, __error_column_titles_9, __error_row_7,
1032 ... __setup, ";", "#"))
1033 ... except ValueError as ve:
1034 ... print(ve)
1035 Too many columns in ['x', 'y'], should be 1.
1037 >>> try:
1038 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
1039 ... "", "#", footer_comments=__err_cmt_1))
1040 ... except ValueError as ve:
1041 ... print(ve)
1042 Invalid separator ''.
1044 >>> try:
1045 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
1046 ... "x", "#", footer_comments=1))
1047 ... except TypeError as te:
1048 ... print(str(te)[:70])
1049 footer_comments should be an instance of typing.Iterable or a callable
1051 >>> try:
1052 ... list(csv_write(dd, lambda x: x, __get_row, __setup,
1053 ... "x", "#", footer_bottom_comments=1))
1054 ... except TypeError as te:
1055 ... print(str(te)[:70])
1056 footer_bottom_comments should be an instance of typing.Iterable or a c
1058 >>> ddx = [{"a": 1, "c": 2}, None,
1059 ... {"a": 4, "d": 12, "b": 3}, {}]
1060 >>> def __error_row_9(_, __):
1061 ... return ("1", "2", "3", "4")
1062 >>> def __error_row_10(_):
1063 ... __error_row_9(1, 2)
1065 >>> try:
1066 ... list(csv_write(ddx, __error_row_10,
1067 ... __error_row_9, lambda x: x, ";", "#"))
1068 ... except TypeError as te:
1069 ... print(te)
1070 'NoneType' object is not iterable
1071 """
1072 if not (isinstance(column_titles, Iterable) or callable(column_titles)):
1073 raise type_error(column_titles, "column_titles", Iterable, call=True)
1074 if not callable(get_row):
1075 raise type_error(get_row, "get_row", call=True)
1076 if not callable(setup):
1077 raise type_error(setup, "setup", call=True)
1078 if str.__len__(separator) <= 0:
1079 raise ValueError(f"Invalid separator {separator!r}.")
1080 forbidden_marker: Final[set[str]] = set(NEWLINE)
1081 forbidden_marker.add(separator)
1082 if comment_start is not None:
1083 if (str.__len__(comment_start) <= 0) or (
1084 str.strip(comment_start) != comment_start) or (
1085 comment_start in separator):
1086 raise ValueError(f"Invalid comment start: {comment_start!r}.")
1087 forbidden_marker.add(comment_start)
1088 if (header_comments is not None) and (not (isinstance(
1089 header_comments, Iterable) or callable(header_comments))):
1090 raise type_error(
1091 header_comments, "header_comments", Iterable, call=True)
1092 if (footer_comments is not None) and (not (isinstance(
1093 footer_comments, Iterable) or callable(footer_comments))):
1094 raise type_error(
1095 footer_comments, "footer_comments", Iterable, call=True)
1096 if (footer_bottom_comments is not None) and (not (isinstance(
1097 footer_bottom_comments, Iterable) or callable(
1098 footer_bottom_comments))):
1099 raise type_error(footer_bottom_comments,
1100 "footer_bottom_comments", Iterable, call=True)
1102 data = reiterable(data, True) # we can iterate over the data twice
1103 setting: Final[S] = setup(data)
1104 forbidden: Final[list[str]] = sorted(forbidden_marker)
1106 # first put header comments
1107 if (comment_start is not None) and (header_comments is not None):
1108 yield from __print_comments(
1109 header_comments(setting) if callable(header_comments)
1110 else header_comments, comment_start, "header", False)
1112 # now process the column titles
1113 collected: list[str] = list(
1114 column_titles(setting) if callable(column_titles) else column_titles)
1115 col_count: Final[int] = list.__len__(collected)
1116 if col_count <= 0:
1117 raise ValueError("Cannot have zero columns.")
1118 for i, col in enumerate(collected):
1119 collected[i] = xcol = str.strip(col)
1120 if (str.__len__(xcol) <= 0) or any(map(xcol.__contains__, forbidden)):
1121 raise ValueError(f"Invalid column title {col!r}, must neither be"
1122 f" empty nor contain any of {forbidden!r}.")
1123 if set.__len__(set(collected)) != col_count:
1124 raise ValueError(f"Cannot have duplicated columns: {collected!r}.")
1125 yield separator.join(collected)
1127 # now do the single rows
1128 data = unpack_reiterable(data) # only need to go over the data once more
1129 for element in data:
1130 if element is None:
1131 raise type_error(element, "data element", object)
1132 collected.clear()
1133 collected.extend(get_row(setting, element))
1134 list_len: int = list.__len__(collected)
1135 if list_len > col_count:
1136 raise ValueError(
1137 f"Too many columns in {collected!r}, should be {col_count}.")
1138 last_non_empty: int = -1
1139 for i, col in enumerate(collected):
1140 collected[i] = xcol = str.strip(col)
1141 if any(map(xcol.__contains__, forbidden)):
1142 raise ValueError(f"Invalid column value {col!r}, cannot "
1143 f"contain any of {forbidden!r}.")
1144 if str.__len__(xcol) > 0:
1145 last_non_empty = i + 1
1146 if last_non_empty < list_len:
1147 if last_non_empty <= 0:
1148 if col_count <= 1:
1149 raise ValueError(
1150 f"Cannot have empty row in a single-column format, "
1151 f"but got {collected!r}.")
1152 yield separator
1153 continue
1154 del collected[last_non_empty:]
1155 yield separator.join(collected)
1157 # finally put footer comments
1158 if comment_start is not None:
1159 empty_next: bool = False
1160 if footer_comments is not None:
1161 for c in __print_comments(footer_comments(setting) if callable(
1162 footer_comments) else footer_comments, comment_start,
1163 "footer", False):
1164 yield c
1165 empty_next = True
1166 if footer_bottom_comments is not None:
1167 yield from __print_comments(
1168 footer_bottom_comments(setting) if callable(
1169 footer_bottom_comments) else footer_bottom_comments,
1170 comment_start, "footer bottom", empty_next)
1173def csv_str_or_none(data: list[str | None] | None,
1174 index: int | None) -> str | None:
1175 """
1176 Get a string or `None` from a data row.
1178 This function is a shortcut for when data elements or columns are
1179 optional. If `index` is `None` or outside of the valid index range of the
1180 list `data`, then `None` is returned. If `data` itself is `None` or the
1181 element at index `index` is the empty string, then `None` is returned.
1182 Only if `data` and `index` are both not `None` and `index` is a valid
1183 index into `data` and the element at index `index` in `data` is not the
1184 empty string, then this element is returned. In other words, this is a
1185 very tolerant function to handle optional data and to return `None` if the
1186 data is not present. The function :func:`csv_val_or_none` further extends
1187 this function by converting the data to another data type if it is
1188 present.
1190 :param data: the data
1191 :param index: the index, if any
1192 :returns: the string or nothing
1194 >>> ddd = ["a", "b", "", "d"]
1195 >>> print(csv_str_or_none(ddd, 0))
1196 a
1197 >>> print(csv_str_or_none(ddd, 1))
1198 b
1199 >>> print(csv_str_or_none(ddd, 2))
1200 None
1201 >>> print(csv_str_or_none(ddd, 3))
1202 d
1203 >>> print(csv_str_or_none(ddd, None))
1204 None
1205 >>> print(csv_str_or_none(ddd, 10))
1206 None
1207 >>> print(csv_str_or_none(ddd, -1))
1208 None
1209 >>> print(csv_str_or_none(None, 0))
1210 None
1211 """
1212 if (index is None) or (data is None):
1213 return None
1214 if 0 <= index <= list.__len__(data):
1215 d: str = data[index]
1216 return None if (d is None) or (str.__len__(d) <= 0) else d
1217 return None
1220#: a type variable for :func:`csv_val_or_none`.
1221U = TypeVar("U")
1224def csv_val_or_none(data: list[str | None] | None, index: int | None,
1225 conv: Callable[[str], U]) -> U | None:
1226 """
1227 Get a value or `None`.
1229 See :func:`csv_str_or_none` allows us to extract an optional data element
1230 from a CSV row and get `None` if the element is not present or if the
1231 `index` is `None` or outside of the valid range. In case the data is
1232 present and not the empty string, then the function `conv` is invoked to
1233 convert it to another value. Otherwise, `None` is returned.
1235 :param data: the data
1236 :param index: the index
1237 :param conv: the conversation function
1238 :returns: the object
1240 >>> ddd = ["11", "22", "", "33"]
1241 >>> print(csv_val_or_none(ddd, 0, int))
1242 11
1243 >>> print(csv_val_or_none(ddd, 1, int))
1244 22
1245 >>> print(csv_val_or_none(ddd, 2, int))
1246 None
1247 >>> print(csv_val_or_none(ddd, 3, int))
1248 33
1249 >>> print(csv_val_or_none(ddd, None, int))
1250 None
1251 """
1252 t: Final[str | None] = csv_str_or_none(data, index)
1253 return None if t is None else conv(t)
1256def csv_column(columns: dict[str, int], key: str,
1257 remove_col: bool = True) -> int:
1258 """
1259 Get the index of a CSV column.
1261 This function will extract the index of a column from a column description
1262 map. The index will be checked whether it is in a valid range and
1263 returned. If no column fitting to `key` exists, this function will throw a
1264 `KeyError`. If `remove_col` is `True` and a column fitting to `key`
1265 exists, then this column will be deleted from `columns`.
1267 :param columns: the columns set
1268 :param key: the key
1269 :param remove_col: should we remove the column?
1270 :returns: the column
1271 :raises TypeError: if any of the parameters is not of the prescribed type
1272 :raises ValueError: if the column or key are invalid
1273 :raises KeyError: if no column of the name `key` eixists
1275 >>> csv_column({"a": 5}, "a")
1276 5
1278 >>> cols = {"a": 5, "b": 7}
1279 >>> csv_column(cols, "a", False)
1280 5
1281 >>> cols
1282 {'a': 5, 'b': 7}
1283 >>> csv_column(cols, "a", True)
1284 5
1285 >>> cols
1286 {'b': 7}
1288 >>> try:
1289 ... csv_column({"a": 5}, "b")
1290 ... except KeyError as ke:
1291 ... print(ke)
1292 'b'
1294 >>> try:
1295 ... csv_column({"a": 5}, "a", "3")
1296 ... except TypeError as te:
1297 ... print(te)
1298 remove_col should be an instance of bool but is str, namely '3'.
1300 >>> try:
1301 ... csv_column(None, "b")
1302 ... except TypeError as te:
1303 ... print(str(te)[:50])
1304 descriptor '__getitem__' for 'dict' objects doesn'
1306 >>> try:
1307 ... csv_column({"a": 5}, 1)
1308 ... except TypeError as te:
1309 ... print(te)
1310 descriptor '__len__' requires a 'str' object but received a 'int'
1312 >>> try:
1313 ... csv_column({"a": -1}, "a")
1314 ... except ValueError as ve:
1315 ... print(ve)
1316 a=-1 is invalid, must be in 0..1000000.
1318 >>> try:
1319 ... csv_column({"a": -1}, "")
1320 ... except ValueError as ve:
1321 ... print(ve)
1322 Invalid key ''.
1323 """
1324 if str.__len__(key) <= 0:
1325 raise ValueError(f"Invalid key {key!r}.")
1326 if not isinstance(remove_col, bool):
1327 raise type_error(remove_col, "remove_col", bool)
1328 res: Final[int] = check_int_range(dict.__getitem__(
1329 columns, key), key, 0, 1_000_000)
1330 if remove_col:
1331 dict.__delitem__(columns, key)
1332 return res
1335def csv_column_or_none(columns: dict[str, int] | None = None,
1336 key: str | None = None,
1337 remove_col: bool = True) -> int | None:
1338 """
1339 Get an optional CSV column index.
1341 This function will extract the index of a column from a column description
1342 map. The index will be checked whether it is in a valid range and
1343 returned. If no column fitting to `key` exists, this function returns
1344 `None`. If `remove_col` is `True` and a column fitting to `key` exists,
1345 then this column will be deleted from `columns`.
1347 :param columns: the columns
1348 :param key: the key
1349 :param remove_col: should we remove the column?
1350 :returns: the column, or `None` if none was found
1351 :raises TypeError: if any of the parameters is not of the prescribed type
1352 :raises ValueError: if the column or key are invalid
1354 >>> csv_column_or_none({"a": 5}, "a")
1355 5
1357 >>> cols = {"a": 5, "b": 7}
1358 >>> csv_column_or_none(cols, "a", False)
1359 5
1360 >>> cols
1361 {'a': 5, 'b': 7}
1362 >>> csv_column_or_none(cols, "a", True)
1363 5
1364 >>> cols
1365 {'b': 7}
1367 >>> try:
1368 ... csv_column_or_none({"a": 5}, "a", "3")
1369 ... except TypeError as te:
1370 ... print(te)
1371 remove_col should be an instance of bool but is str, namely '3'.
1373 >>> print(csv_column_or_none({"a": 5}, "b"))
1374 None
1376 >>> print(csv_column_or_none(None, "b"))
1377 None
1379 >>> print(csv_column_or_none({"a": 5}, None))
1380 None
1382 >>> print(csv_column_or_none({"a": 5}, ""))
1383 None
1385 >>> try:
1386 ... csv_column({"a": 5}, 1)
1387 ... except TypeError as te:
1388 ... print(te)
1389 descriptor '__len__' requires a 'str' object but received a 'int'
1391 >>> try:
1392 ... csv_column({"a": -1}, "a")
1393 ... except ValueError as ve:
1394 ... print(ve)
1395 a=-1 is invalid, must be in 0..1000000.
1396 """
1397 if not isinstance(remove_col, bool):
1398 raise type_error(remove_col, "remove_col", bool)
1399 if (key is None) or (columns is None) or (str.__len__(key) <= 0):
1400 return None
1401 res: Final[int | None] = dict.get(columns, key)
1402 if res is None:
1403 return None
1404 check_int_range(res, key, 0, 1_000_000)
1405 if remove_col:
1406 dict.__delitem__(columns, key)
1407 return res
1410def csv_select_scope(
1411 conv: Callable[[dict[str, int]], U],
1412 columns: dict[str, int],
1413 scope: str | None = None,
1414 additional: Iterable[tuple[str, int]] = (),
1415 skip_orig_key: Callable[[str], bool] = lambda _: False,
1416 skip_final_key: Callable[[str], bool] = lambda _: False,
1417 skip_col: Callable[[int], bool] = lambda _: False,
1418 include_scope: bool = True,
1419 remove_cols: bool = True) -> U:
1420 """
1421 Get all the columns of a given scope and pass them to the function `conv`.
1423 This function is intended for selecting some keys from a column set and
1424 pass them as parameters to a constructor of a CSV reader. It can do this
1425 selection based on a `scope` prefix which is then removed from the column
1426 names before passing them into the constructor. If no column matches, this
1427 function throws a :class:`ValueError`.
1428 All columns that are passed on to `conv` are deleted from `columns` if
1429 `remove_cols == True`, which is the default.
1431 :param conv: the function to which the selected columns should be passed,
1432 and that creates the return value
1433 :param columns: the existing columns
1434 :param scope: the scope, or `None` or the empty string to select all
1435 columns
1436 :param skip_orig_key: a function that returns `True` for any original,
1437 unchanged key in `columns` that should be ignored and that
1438 returns `False` if the key can be processed normally (i.e., if we can
1439 check if it starts with the given scope and move on)
1440 :param skip_final_key: a function that returns `True` for any key in
1441 `columns` that would fall into the right scope but that should still
1442 be ignored. This function receives the key without the scope prefix.
1443 :param skip_col: any column that should be ignored
1444 :param additional: the additional columns to add *if* some keys/columns
1445 remain after all the transformation and selection
1446 :param include_scope: if scope appears as a lone column, should we
1447 include it?
1448 :param remove_cols: should we remove all selected columns?
1449 :returns: The result of the function `conv` applied to all matching
1450 columns (and those in `additional` are appended to them)
1451 :raises ValueError: if no columns could be selected
1452 :raises TypeError: if any of the elements passed in is of the wrong type
1454 >>> csv_select_scope(lambda x: x, {
1455 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "")
1456 {'a.x': 1, 'a.y': 2, 'a': 3, 'b': 4, 'b.t': 5}
1458 >>> try:
1459 ... csv_select_scope(print, {"a.x": 1, "a.y": 2}, "v")
1460 ... except ValueError as ve:
1461 ... print(ve)
1462 Did not find sufficient data of scope 'v' in {'a.x': 1, 'a.y': 2}.
1464 >>> try:
1465 ... csv_select_scope(print, {}, "v")
1466 ... except ValueError as ve:
1467 ... print(ve)
1468 Did not find sufficient data of scope 'v' in {}.
1469 """
1470 res: Final[U | None] = csv_select_scope_or_none(
1471 conv, columns, scope, additional, skip_orig_key, skip_final_key,
1472 skip_col, include_scope, remove_cols) \
1473 if dict.__len__(columns) > 0 else None
1474 if res is None:
1475 raise ValueError("Did not find sufficient data of "
1476 f"scope {scope!r} in {columns!r}.")
1477 return res
1480def csv_select_scope_or_none(
1481 conv: Callable[[dict[str, int]], U],
1482 columns: dict[str, int] | None,
1483 scope: str | None = None,
1484 additional: Iterable[tuple[str, int]] = (),
1485 skip_orig_key: Callable[[str], bool] = lambda _: False,
1486 skip_final_key: Callable[[str], bool] = lambda _: False,
1487 skip_col: Callable[[int], bool] = lambda _: False,
1488 include_scope: bool = True,
1489 remove_cols: bool = True) -> U | None:
1490 """
1491 Get all the columns of a given scope and pass them to the function `conv`.
1493 This function is intended for selecting some keys from a column set and
1494 pass them as parameters to a constructor of a CSV reader. It can do this
1495 selection based on a `scope` prefix which is then removed from the column
1496 names before passing them into the constructor. If no column matches, this
1497 function returns `None`.
1498 All columns that are passed on to `conv` are deleted from `columns` if
1499 `remove_cols == True`, which is the default.
1501 :param conv: the function to which the selected columns should be passed,
1502 if any, and that - in this case, returns the return value of this
1503 function
1504 :param columns: the existing columns
1505 :param scope: the scope, or `None` or the empty string to select all
1506 columns
1507 :param skip_orig_key: a function that returns `True` for any original,
1508 unchanged key in `columns` that should be ignored and that
1509 returns `False` if the key can be processed normally (i.e., if we can
1510 check if it starts with the given scope and move on)
1511 :param skip_final_key: a function that returns `True` for any key in
1512 `columns` that would fall into the right scope but that should still
1513 be ignored. This function receives the key without the scope prefix.
1514 :param skip_col: any column that should be ignored
1515 :param additional: the additional columns to add *if* some keys/columns
1516 remain after all the transformation and selection
1517 :param include_scope: if scope appears as a lone column, should we
1518 include it?
1519 :param remove_cols: should we remove all selected columns?
1520 :returns: `None` if no keys fall into the provided scope does not have any
1521 keys matching it in `columns`. The result of `conv` otherwise, i.e.,
1522 if there are matching columns, these are selected (and those in
1523 `additional` are appended to them) and these are then passed to `conv`
1524 and the result of `conv` is returned
1526 >>> csv_select_scope_or_none(print, {
1527 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a")
1528 {'x': 1, 'y': 2, 'a': 3}
1530 >>> exa1 = {"a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}
1531 >>> csv_select_scope_or_none(print, exa1, "a", remove_cols=False)
1532 {'x': 1, 'y': 2, 'a': 3}
1533 >>> exa1
1534 {'a.x': 1, 'a.y': 2, 'a': 3, 'b': 4, 'b.t': 5}
1535 >>> csv_select_scope_or_none(print, exa1, "a", remove_cols=True)
1536 {'x': 1, 'y': 2, 'a': 3}
1537 >>> exa1
1538 {'b': 4, 'b.t': 5}
1539 >>> csv_select_scope_or_none(print, exa1, "b", remove_cols=True)
1540 {'b': 4, 't': 5}
1541 >>> exa1
1542 {}
1544 >>> csv_select_scope_or_none(print, {
1545 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "")
1546 {'a.x': 1, 'a.y': 2, 'a': 3, 'b': 4, 'b.t': 5}
1548 >>> csv_select_scope_or_none(print, {
1549 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, None)
1550 {'a.x': 1, 'a.y': 2, 'a': 3, 'b': 4, 'b.t': 5}
1552 >>> csv_select_scope_or_none(print, {
1553 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1554 ... include_scope=False)
1555 {'x': 1, 'y': 2}
1557 >>> csv_select_scope_or_none(print, {
1558 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b")
1559 {'b': 4, 't': 5}
1561 >>> csv_select_scope_or_none(print, {
1562 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b",
1563 ... additional=(('z', 23), ('v', 45)))
1564 {'b': 4, 't': 5, 'z': 23, 'v': 45}
1566 >>> csv_select_scope_or_none(print, {
1567 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b",
1568 ... additional=(('t', 23), ('v', 45)))
1569 {'b': 4, 't': 5, 'v': 45}
1571 >>> csv_select_scope_or_none(print, {
1572 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1573 ... additional=(('x', 44), ('v', 45)))
1574 {'x': 1, 'y': 2, 'a': 3, 'v': 45}
1576 >>> csv_select_scope_or_none(print, {
1577 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b",
1578 ... additional=(('z', 23), ('v', 45)),
1579 ... skip_col=lambda c: c == 23)
1580 {'b': 4, 't': 5, 'v': 45}
1582 >>> csv_select_scope_or_none(print, {
1583 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b",
1584 ... additional=(('z', 23), ('v', 45)),
1585 ... skip_orig_key=lambda ok: ok == "b.t")
1586 {'b': 4, 'z': 23, 'v': 45}
1588 >>> csv_select_scope_or_none(print, {
1589 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "b",
1590 ... additional=(('z', 23), ('v', 45)),
1591 ... skip_final_key=lambda fk: fk == "z")
1592 {'b': 4, 't': 5, 'v': 45}
1594 >>> print(csv_select_scope_or_none(print, {}, "a"))
1595 None
1597 >>> print(csv_select_scope_or_none(print, {}, None))
1598 None
1600 >>> print(csv_select_scope_or_none(print, None, None))
1601 None
1603 >>> print(csv_select_scope_or_none(print, {"a.x": 45}, "a",
1604 ... skip_col=lambda c: c == 45))
1605 None
1607 >>> try:
1608 ... csv_select_scope_or_none(None, {
1609 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a")
1610 ... except TypeError as te:
1611 ... print(te)
1612 conv should be a callable but is None.
1614 >>> try:
1615 ... csv_select_scope_or_none(print, {
1616 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1617 ... remove_cols=1)
1618 ... except TypeError as te:
1619 ... print(te)
1620 remove_cols should be an instance of bool but is int, namely 1.
1622 >>> try:
1623 ... csv_select_scope_or_none("x", {
1624 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a")
1625 ... except TypeError as te:
1626 ... print(te)
1627 conv should be a callable but is str, namely 'x'.
1629 >>> try:
1630 ... csv_select_scope_or_none(print, "x", "a")
1631 ... except TypeError as te:
1632 ... print(te)
1633 descriptor '__len__' requires a 'dict' object but received a 'str'
1635 >>> try:
1636 ... csv_select_scope_or_none(print, {
1637 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, int)
1638 ... except TypeError as te:
1639 ... print(te)
1640 descriptor '__len__' requires a 'str' object but received a 'type'
1642 >>> try:
1643 ... csv_select_scope_or_none(print, {
1644 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1645 ... additional=2)
1646 ... except TypeError as te:
1647 ... print(str(te)[:-7])
1648 additional should be an instance of typing.Iterable but is int, na
1650 >>> try:
1651 ... csv_select_scope_or_none(print, {
1652 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1653 ... additional=((1, 2), ))
1654 ... except TypeError as te:
1655 ... print(te)
1656 descriptor '__len__' requires a 'str' object but received a 'int'
1658 >>> try:
1659 ... csv_select_scope_or_none(print, {
1660 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1661 ... additional=(None, ))
1662 ... except TypeError as te:
1663 ... print(te)
1664 cannot unpack non-iterable NoneType object
1666 >>> try:
1667 ... csv_select_scope_or_none(print, {
1668 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1669 ... additional=(("yx", "a"), ))
1670 ... except TypeError as te:
1671 ... print(te)
1672 yx should be an instance of int but is str, namely 'a'.
1674 >>> try:
1675 ... csv_select_scope_or_none(print, {
1676 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1677 ... additional=(("yx", -2), ))
1678 ... except ValueError as ve:
1679 ... print(ve)
1680 yx=-2 is invalid, must be in 0..1000000.
1682 >>> try:
1683 ... csv_select_scope_or_none(print, {
1684 ... "a.x": 1, "a.y": 2, "a": 3, "a.b": -4, "b.t": 5}, "a")
1685 ... except ValueError as ve:
1686 ... print(ve)
1687 a.b=-4 is invalid, must be in 0..1000000.
1689 >>> try:
1690 ... csv_select_scope_or_none(print, {
1691 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1692 ... skip_col=None)
1693 ... except TypeError as te:
1694 ... print(te)
1695 skip_col should be a callable but is None.
1697 >>> try:
1698 ... csv_select_scope_or_none(print, {
1699 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1700 ... skip_orig_key=None)
1701 ... except TypeError as te:
1702 ... print(te)
1703 skip_orig_key should be a callable but is None.
1705 >>> try:
1706 ... csv_select_scope_or_none(print, {
1707 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1708 ... skip_final_key=None)
1709 ... except TypeError as te:
1710 ... print(te)
1711 skip_final_key should be a callable but is None.
1713 >>> try:
1714 ... csv_select_scope(print, {
1715 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1716 ... include_scope=3)
1717 ... except TypeError as te:
1718 ... print(te)
1719 include_scope should be an instance of bool but is int, namely 3.
1721 >>> try:
1722 ... csv_select_scope_or_none(print, {
1723 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, 4)
1724 ... except TypeError as te:
1725 ... print(te)
1726 descriptor '__len__' requires a 'str' object but received a 'int'
1728 >>> try:
1729 ... csv_select_scope_or_none(print, 11)
1730 ... except TypeError as te:
1731 ... print(te)
1732 descriptor '__len__' requires a 'dict' object but received a 'int'
1734 >>> try:
1735 ... csv_select_scope_or_none(print, {
1736 ... "a.x": 1, "a.y": 2, "a": 3, "b": 4, "b.t": 5}, "a",
1737 ... additional=(("", 2), ))
1738 ... except ValueError as ve:
1739 ... print(ve)
1740 Invalid additional column ''.
1741 """
1742 if not callable(conv):
1743 raise type_error(conv, "conv", call=True)
1744 if not callable(skip_orig_key):
1745 raise type_error(skip_orig_key, "skip_orig_key", call=True)
1746 if not callable(skip_final_key):
1747 raise type_error(skip_final_key, "skip_final_key", call=True)
1748 if not isinstance(additional, Iterable):
1749 raise type_error(additional, "additional", Iterable)
1750 if not isinstance(include_scope, bool):
1751 raise type_error(include_scope, "include_scope", bool)
1752 if not callable(skip_col):
1753 raise type_error(skip_col, "skip_col", call=True)
1754 if not isinstance(remove_cols, bool):
1755 raise type_error(remove_cols, "remove_cols", bool)
1757 if (columns is None) or (dict.__len__(columns) <= 0):
1758 return None
1759 selection: Final[list[tuple[str, str, int]]] = [
1760 (k, k, v) for k, v in columns.items()
1761 if not (skip_orig_key(k) or skip_col(v))]
1762 sel_len: Final[int] = list.__len__(selection)
1763 if sel_len <= 0:
1764 return None
1766 if (scope is not None) and (str.__len__(scope) > 0):
1767 use_scope: Final[str] = f"{scope}{SCOPE_SEPARATOR}"
1768 usl: Final[int] = str.__len__(use_scope)
1769 for i in range(sel_len - 1, -1, -1):
1770 k, _, v = selection[i]
1771 if str.startswith(k, use_scope):
1772 use_key = k[usl:]
1773 if not skip_final_key(use_key):
1774 list.__setitem__(selection, i, (k, use_key, v))
1775 continue
1776 elif include_scope and (k == scope):
1777 if not skip_final_key(k):
1778 continue
1779 list.__delitem__(selection, i)
1781 if list.__len__(selection) <= 0:
1782 return None
1784 if remove_cols:
1785 for kv in selection:
1786 dict.__delitem__(columns, kv[0])
1788 subset: Final[dict[str, int]] = {
1789 kv[1]: check_int_range(
1790 kv[2], kv[0], 0, 1_000_000) for kv in selection}
1792 for kkk, vvv in additional:
1793 if str.__len__(kkk) <= 0:
1794 raise ValueError(f"Invalid additional column {kkk!r}.")
1795 if skip_final_key(kkk) or skip_col(vvv):
1796 continue
1797 if kkk not in subset:
1798 subset[kkk] = check_int_range(vvv, kkk, 0, 1_000_000)
1799 return conv(subset)
1802class CsvReader[T]:
1803 """
1804 A base class for CSV readers.
1806 Using this class and its :meth:`read` class method provides for a more
1807 elegant way to construct nested and combined CSV formats compared to
1808 creating classes and handing their methods to :func:`csv_read`.
1810 >>> class R(CsvReader):
1811 ... def __init__(self, columns: dict[str, int]) -> None:
1812 ... super().__init__(columns)
1813 ... self.cols = columns
1814 ... def parse_row(self, row: list[str]) -> dict:
1815 ... return {x: row[y] for x, y in self.cols.items()}
1817 >>> text = ["a;b;c;d", "# test", " 1; 2;3;4", " 5 ;6 ", ";8;;9",
1818 ... "", "10", "# 11;12"]
1820 >>> for p in R.read(text):
1821 ... print(p)
1822 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
1823 {'a': '5', 'b': '6', 'c': '', 'd': ''}
1824 {'a': '', 'b': '8', 'c': '', 'd': '9'}
1825 {'a': '10', 'b': '', 'c': '', 'd': ''}
1827 >>> text = ["a,b,c,d", "v test", " 1, 2,3,4", " 5 ,6 ", ",8,,9",
1828 ... "", "10", "v 11,12"]
1830 >>> for p in R.read(text, separator=',', comment_start='v'):
1831 ... print(p)
1832 {'a': '1', 'b': '2', 'c': '3', 'd': '4'}
1833 {'a': '5', 'b': '6', 'c': '', 'd': ''}
1834 {'a': '', 'b': '8', 'c': '', 'd': '9'}
1835 {'a': '10', 'b': '', 'c': '', 'd': ''}
1837 >>> class S(CsvReader):
1838 ... def __init__(self, columns: dict[str, int], add: str) -> None:
1839 ... super().__init__(columns)
1840 ... self.cols = columns
1841 ... self.s = add
1842 ... def parse_row(self, row: list[str]) -> dict:
1843 ... return {x: self.s + row[y] for x, y in self.cols.items()}
1845 >>> text = ["a;b;c;d", "# test", " 1; 2;3;4", " 5 ;6 ", ";8;;9",
1846 ... "", "10", "# 11;12"]
1848 >>> for p in S.read(text, add="b"):
1849 ... print(p)
1850 {'a': 'b1', 'b': 'b2', 'c': 'b3', 'd': 'b4'}
1851 {'a': 'b5', 'b': 'b6', 'c': 'b', 'd': 'b'}
1852 {'a': 'b', 'b': 'b8', 'c': 'b', 'd': 'b9'}
1853 {'a': 'b10', 'b': 'b', 'c': 'b', 'd': 'b'}
1855 >>> ccc = S({"a": 1}, add="x")
1856 >>> print(ccc.parse_optional_row(None))
1857 None
1858 >>> print(S.parse_optional_row(None, None))
1859 None
1860 >>> print((ccc).parse_optional_row(["x", "y"]))
1861 {'a': 'xy'}
1863 >>> try:
1864 ... CsvReader("x")
1865 ... except TypeError as te:
1866 ... print(te)
1867 columns should be an instance of dict but is str, namely 'x'.
1869 >>> try:
1870 ... CsvReader({"a": 1}).parse_row(["a"])
1871 ... except NotImplementedError as nie:
1872 ... print(type(nie))
1873 <class 'NotImplementedError'>
1874 """
1876 def __init__(self, columns: dict[str, int]) -> None:
1877 """
1878 Create the CSV reader.
1880 :param columns: the columns
1881 :raises TypeError: if `columns` is not a :class:`dict`
1882 """
1883 super().__init__()
1884 if not isinstance(columns, dict):
1885 raise type_error(columns, "columns", dict)
1887 def parse_row(self, data: list[str]) -> T:
1888 """
1889 Parse a row of data.
1891 :param data: the data row
1892 :returns: the object representing the row
1893 :raises NotImplementedError: because it must be overridden
1894 :raises ValueError: should raise a :class:`ValueError` if the row is
1895 incomplete or invalid
1896 """
1897 raise NotImplementedError
1899 def parse_optional_row(self, data: list[str] | None) -> T | None:
1900 """
1901 Parse a row of data that may be incomplete or empty.
1903 The default implementation of this method returns `None` if the data
1904 row is `None`, or if `self` is `None`, which should never happen.
1905 Otherwise, it calls :meth:`parse_row`, which will probably raise a
1906 :class:`ValueError`.
1908 :param data: the row of data that may be empty
1909 :returns: an object constructed from the partial row, if possible,
1910 or `None`
1911 """
1912 if (self is None) or (data is None):
1913 return None
1914 return self.parse_row(data)
1916 @classmethod
1917 def read(cls: type["CsvReader"], rows: Iterable[str],
1918 separator: str = CSV_SEPARATOR,
1919 comment_start: str | None = COMMENT_START,
1920 **kwargs) -> Generator[T, None, None]:
1921 """
1922 Parse a stream of CSV data.
1924 This class method creates a single new instance of `cls` and passes it
1925 the column names/indices as well as any additional named arguments of
1926 this method into the constructor. It then uses the method
1927 :meth:`parse_row` of the class to parse the row data to generate the
1928 output stream.
1930 It offers a more convenient wrapper around :func:`csv_read` for cases
1931 where it makes more sense to implement the parsing functionality in a
1932 class.
1934 :param rows: the rows of strings with CSV data
1935 :param separator: the separator character
1936 :param comment_start: the comment start character
1937 """
1938 def __creator(y: dict[str, int], __c=cls, # pylint: disable=W0102
1939 __x=kwargs) -> "CsvReader": # noqa # type: ignore
1940 return cls(y, **__x) # noqa # type: ignore
1942 yield from csv_read(rows=rows,
1943 setup=__creator,
1944 parse_row=cls.parse_row, # type: ignore
1945 separator=separator,
1946 comment_start=comment_start)
1949class CsvWriter[T]:
1950 """
1951 A base class for structured CSV writers.
1953 >>> class W(CsvWriter):
1954 ... def __init__(self, data: Iterable[dict[str, int]],
1955 ... scope: str | None = None) -> None:
1956 ... super().__init__(data, scope)
1957 ... self.cols = sorted({dkey for datarow in data
1958 ... for dkey in datarow})
1959 ... def get_column_titles(self) -> Iterable[str]:
1960 ... return self.cols
1961 ... def get_row(self, row: dict[str, int]) -> Iterable[str]:
1962 ... return map(str, (row.get(key, "") for key in self.cols))
1963 ... def get_header_comments(self) -> list[str]:
1964 ... return ["This is a header comment.", " We have two of it. "]
1965 ... def get_footer_comments(self) -> list[str]:
1966 ... return [" This is a footer comment."]
1968 >>> dd = [{"a": 1, "c": 2}, {"b": 6, "c": 8},
1969 ... {"a": 4, "d": 12, "b": 3}, {}]
1971 >>> for p in W.write(dd):
1972 ... print(p[:p.index(", version")] if ", version" in p else p)
1973 # This is a header comment.
1974 # We have two of it.
1975 a;b;c;d
1976 1;;2
1977 ;6;8
1978 4;3;;12
1979 ;
1980 # This is a footer comment.
1981 #
1982 # This CSV output has been created using the versatile CSV API of \
1983pycommons.io.csv
1984 # You can find pycommons at https://thomasweise.github.io/pycommons.
1986 >>> for p in W.write(dd):
1987 ... print(p[:p.index(", version")] if ", version" in p else p)
1988 # This is a header comment.
1989 # We have two of it.
1990 a;b;c;d
1991 1;;2
1992 ;6;8
1993 4;3;;12
1994 ;
1995 # This is a footer comment.
1996 #
1997 # This CSV output has been created using the versatile CSV API of \
1998pycommons.io.csv
1999 # You can find pycommons at https://thomasweise.github.io/pycommons.
2001 Now we do the same test, but use the iterator over the data only.
2002 >>> for p in W.write(iter(dd)):
2003 ... print(p[:p.index(", version")] if ", version" in p else p)
2004 # This is a header comment.
2005 # We have two of it.
2006 a;b;c;d
2007 1;;2
2008 ;6;8
2009 4;3;;12
2010 ;
2011 # This is a footer comment.
2012 #
2013 # This CSV output has been created using the versatile CSV API of \
2014pycommons.io.csv
2015 # You can find pycommons at https://thomasweise.github.io/pycommons.
2017 >>> class W2(CsvWriter):
2018 ... def __init__(self, data: Iterable[dict[str, int]],
2019 ... scope: str | None = None) -> None:
2020 ... super().__init__(data, scope)
2021 ... self.cols = ("a", "b", "c", "d")
2022 ... next(iter(data))
2023 ... def get_column_titles(self) -> Iterable[str]:
2024 ... return self.cols if self.scope is None else [
2025 ... f"{self.scope}.{r}" for r in self.cols]
2026 ... def get_row(self, row: dict[str, int]) -> Iterable[str]:
2027 ... return map(str, (row.get(key, "") for key in self.cols))
2028 ... def get_footer_bottom_comments(self) -> Iterable[str] | None:
2029 ... return ["Bla!"]
2031 >>> for p in W2.write(iter(dd), separator="@", comment_start="B"):
2032 ... print(p)
2033 a@b@c@d
2034 1@@2
2035 @6@8
2036 4@3@@12
2037 @
2038 B Bla!
2040 >>> for p in W2.write(iter(dd), scope="k", separator="@",
2041 ... comment_start="B"):
2042 ... print(p)
2043 k.a@k.b@k.c@k.d
2044 1@@2
2045 @6@8
2046 4@3@@12
2047 @
2048 B Bla!
2050 >>> ";".join(W2(dd).get_optional_row(None))
2051 ';;;'
2052 >>> ";".join(W2(dd).get_optional_row(dd[0]))
2053 '1;;2;'
2055 >>> try:
2056 ... CsvWriter(1, None)
2057 ... except TypeError as te:
2058 ... print(te)
2059 data should be an instance of typing.Iterable but is int, namely 1.
2061 >>> try:
2062 ... CsvWriter([], 1)
2063 ... except TypeError as te:
2064 ... print(te)
2065 descriptor 'strip' for 'str' objects doesn't apply to a 'int' object
2067 >>> try:
2068 ... CsvWriter([], "x x")
2069 ... except ValueError as ve:
2070 ... print(ve)
2071 invalid scope 'x x'
2073 >>> try:
2074 ... CsvWriter([], " x")
2075 ... except ValueError as ve:
2076 ... print(ve)
2077 invalid scope ' x'
2079 >>> try:
2080 ... CsvWriter([]).get_row("x")
2081 ... except NotImplementedError as nie:
2082 ... print(type(nie))
2083 <class 'NotImplementedError'>
2085 >>> try:
2086 ... CsvWriter([]).get_column_titles()
2087 ... except NotImplementedError as nie:
2088 ... print(type(nie))
2089 <class 'NotImplementedError'>
2090 """
2092 def __init__(self, data: Iterable[T],
2093 scope: str | None = None) -> None:
2094 """
2095 Initialize the csv writer.
2097 :param data: the data to be written
2098 :param scope: the prefix to be pre-pended to all columns
2099 :raises TypeError: if `data` is not an `Iterable` or if `scope` is
2100 neither `None` nor a string
2101 :raises ValueError: if `scope` is not `None` but: an empty string,
2102 becomes an empty string after stripping, or contains any
2103 whitespace or newline character
2104 """
2105 super().__init__()
2106 if not isinstance(data, Iterable):
2107 raise type_error(data, "data", Iterable)
2108 if (scope is not None) and ((str.strip(scope) != scope) or (
2109 str.__len__(scope) <= 0) or (any(map(
2110 scope.__contains__, WHITESPACE_OR_NEWLINE)))):
2111 raise ValueError(f"invalid scope {scope!r}")
2112 #: the optional scope
2113 self.scope: Final[str | None] = scope
2115 def get_column_titles(self) -> Iterable[str]:
2116 """
2117 Get the column titles.
2119 :returns: the column titles
2120 """
2121 raise NotImplementedError
2123 def get_optional_row(self, data: T | None) -> Iterable[str]:
2124 """
2125 Attach an empty row of the correct shape to the output.
2127 :param data: the data item or `None`
2128 :returns: the optional row data
2129 """
2130 if data is None: # very crude and slow way to create an optional row
2131 return [""] * list.__len__(list(self.get_column_titles()))
2132 return self.get_row(data)
2134 def get_row(self, data: T) -> Iterable[str]:
2135 """
2136 Render a single sample statistics to a CSV row.
2138 :param data: the data sample statistics
2139 :returns: the row iterator
2140 """
2141 raise NotImplementedError
2143 def get_header_comments(self) -> Iterable[str]:
2144 """
2145 Get any possible header comments.
2147 :returns: the iterable of header comments
2148 """
2149 return ()
2151 def get_footer_comments(self) -> Iterable[str]:
2152 """
2153 Get any possible footer comments.
2155 :returns: the footer comments
2156 """
2157 return ()
2159 def get_footer_bottom_comments(self) -> Iterable[str] | None:
2160 """
2161 Get the bottom footer comments.
2163 :returns: an iterator with the bottom comments
2164 """
2165 return pycommons_footer_bottom_comments(self)
2167 @classmethod
2168 def write(
2169 cls: type["CsvWriter"],
2170 data: Iterable[T],
2171 scope: str | None = None,
2172 separator: str = CSV_SEPARATOR,
2173 comment_start: str | None = COMMENT_START,
2174 **kwargs) -> Generator[str, None, None]:
2175 """
2176 Write the CSV data based on the methods provided by the class `cls`.
2178 :param data: the data
2179 :param separator: the CSV separator
2180 :param comment_start: the comment start character
2181 :param scope: the scope, or `None`
2182 :param kwargs: additional arguments to be passed to the constructor
2184 :raises TypeError: if `kwargs` is not `None` but also not a
2185 :class:`dict`
2186 """
2187 def __creator(y: Iterable[T], __c=cls, # pylint: disable=W0102
2188 __s=scope, # noqa # type: ignore
2189 __x=kwargs) -> "CsvWriter": # noqa # type: ignore
2190 return __c(data=y, scope=__s, **__x) # noqa # type: ignore
2192 yield from csv_write(
2193 data=data,
2194 column_titles=cls.get_column_titles, # type: ignore
2195 get_row=cls.get_row, # type: ignore
2196 setup=__creator,
2197 separator=separator,
2198 comment_start=comment_start,
2199 header_comments=cls.get_header_comments, # type: ignore
2200 footer_comments=cls.get_footer_comments, # type: ignore
2201 footer_bottom_comments=cls. # type: ignore
2202 get_footer_bottom_comments) # type: ignore