from __future__ import annotations import csv from collections.abc import Iterable, Iterator from dataclasses import dataclass from typing import NoReturn def _raise_csv_parse_error(error_type: type[Exception], context: str, exc: BaseException) -> NoReturn: if str(exc) == "unexpected end of data": raise error_type(f"{context} CSV row has an unclosed quoted field") from exc raise error_type(f"failed to parse {context} CSV row: {exc}") from exc @dataclass(frozen=True) class CSVParser: error_type: type[Exception] context: str strict: bool = True def iter_rows(self, handle: Iterable[str]) -> Iterator[list[str]]: try: yield from csv.reader(handle, strict=self.strict) except csv.Error as exc: _raise_csv_parse_error(self.error_type, self.context, exc) def new( *, error_type: type[Exception], context: str, strict: bool = True, ) -> CSVParser: return CSVParser(error_type=error_type, context=context, strict=strict)