62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import csv
|
|
import logging
|
|
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 = []
|
|
for n, row in enumerate(rows):
|
|
try:
|
|
records.append(conv(row, headers))
|
|
except ValueError as ex:
|
|
log = logging.getLogger(__name__)
|
|
log.warning(f"Row {n}: bad row: {row}")
|
|
log.debug(f"Reason: {ex}")
|
|
|
|
return records
|
|
|
|
|
|
def csv_as_dicts(
|
|
lines: Iterable[str], types: Sequence[type], headers: Optional[Sequence[str]] = None
|
|
) -> Sequence[Mapping[str, Any]]:
|
|
"""Parse CSV lines into a list of dictionaries."""
|
|
|
|
def _conv(row, hdrs):
|
|
return {name: func(val) for name, func, val in zip(hdrs, types, row)}
|
|
|
|
return convert_csv(lines, _conv, headers)
|
|
|
|
|
|
def read_csv_as_dicts(
|
|
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.
|
|
"""
|
|
with open(filename) as file:
|
|
return csv_as_dicts(file, types, headers)
|
|
|
|
|
|
def csv_as_instances(
|
|
lines: Iterable[str], cls: type, headers: Optional[Sequence[str]] = None
|
|
) -> Sequence[Any]:
|
|
"""Parse CSV lines into a list of of class instances."""
|
|
return convert_csv(lines, lambda row, _: cls.from_row(row), headers)
|
|
|
|
|
|
def read_csv_as_instances(
|
|
filename: str, cls: type, has_headers: bool = True
|
|
) -> Sequence[Any]:
|
|
"""
|
|
Read CSV data into list of instances.
|
|
"""
|
|
with open(filename) as file:
|
|
return csv_as_instances(file, cls, has_headers)
|