This commit is contained in:
Mike Bloy 2023-11-24 12:41:00 -06:00
parent 2e2512ab52
commit f41c496bdb

View File

@ -1,23 +1,33 @@
import csv import csv
from typing import Iterable, Optional, Sequence from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
def convert_csv(
lines: Iterable[str],
conv: Callable[[Sequence[Any], Sequence[str]], Any],
headers: Optional[Sequence[str]] = None,
) -> Sequence[Any]:
rows = csv.reader(lines)
if headers is None:
headers = next(rows)
records = list(map(lambda row: conv(row, headers), rows))
return records
def csv_as_dicts( def csv_as_dicts(
lines: Iterable[str], types: Sequence[type], headers: Optional[Sequence[str]] = None lines: Iterable[str], types: Sequence[type], headers: Optional[Sequence[str]] = None
): ) -> Sequence[Mapping[str, Any]]:
records = [] """Parse CSV lines into a list of dictionaries."""
rows = csv.reader(lines)
if headers is None: def _conv(row, hdrs):
headers = next(rows) return {name: func(val) for name, func, val in zip(hdrs, types, row)}
for row in rows:
record = {name: func(val) for name, func, val in zip(headers, types, row)} return convert_csv(lines, _conv, headers)
records.append(record)
return records
def read_csv_as_dicts( def read_csv_as_dicts(
filename: str, types: Sequence[type], headers: Optional[Sequence[str]] = None filename: str, types: Sequence[type], headers: Optional[Sequence[str]] = None
): ) -> Sequence[Mapping[str, Any]]:
""" """
Read CSV data into list of dictionaries with optional type conversion. Read CSV data into list of dictionaries with optional type conversion.
""" """
@ -25,18 +35,16 @@ def read_csv_as_dicts(
return csv_as_dicts(file, types, headers) return csv_as_dicts(file, types, headers)
def csv_as_instances(lines: Iterable[str], cls: type, has_headers: bool = True): def csv_as_instances(
records = [] lines: Iterable[str], cls: type, headers: Optional[Sequence[str]] = None
rows = csv.reader(lines) ) -> Sequence[Any]:
if has_headers: """Parse CSV lines into a list of of class instances."""
next(rows) return convert_csv(lines, lambda row, _: cls.from_row(row), headers)
for row in rows:
record = cls.from_row(row)
records.append(record)
return records
def read_csv_as_instances(filename: str, cls: type, has_headers: bool = True): def read_csv_as_instances(
filename: str, cls: type, has_headers: bool = True
) -> Sequence[Any]:
""" """
Read CSV data into list of instances. Read CSV data into list of instances.
""" """