feat: keep track of line numbers when making exceptions

tests/test_reader.py

@@ -207,8 +207,8 @@ def test_reader_raises_exception_for_failed_type_coercion(tmp_path: Path) -> Non
     """Test the reader raises an exception for failed type coercion."""
     (tmp_path / "test.txt").write_text(
         "\n".join([
-            "field1\tfield2\tfield3\n",
-            "1\tname\tBOMB\n",
+            "field1\tfield2\tfield3",
+            "1\tname\tBOMB",
         ])
     )
 
@@ -217,7 +217,7 @@ def test_reader_raises_exception_for_failed_type_coercion(tmp_path: Path) -> Non
         pytest.raises(
             DecodeError,
             match=(
-                r"Could not load delimited data line into JSON\-like format\."
+                r"Could not load delimited data into JSON\-like format on line 2\."
                 + r" Built improperly formatted JSON\:"
                 + r" \{\"field1\"\:1\,\"field2\"\:\"name\"\,\"field3\"\:BOMB\}\."
                 + r" Original exception\: JSON is malformed\:"

typeline/_reader.py

@@ -9,6 +9,7 @@
 from dataclasses import fields as fields_of
 from dataclasses import is_dataclass
 from io import TextIOWrapper
+from os import linesep
 from pathlib import Path
 from types import NoneType
 from types import TracebackType
@@ -62,30 +63,30 @@
 
         # Initialize and save internal attributes of this class.
         self._handle: TextIOWrapper = handle
+        self._line_count: int = 0
         self._record_type: type[RecordType] = record_type
         self._comment_prefixes: set[str] = comment_prefixes
 
         # Inspect the record type and save the fields, field names, and field types.
         self._fields: tuple[Field[Any], ...] = fields_of(record_type)
         self._header: list[str] = [field.name for field in self._fields]
         self._field_types: list[type | str | Any] = [field.type for field in self._fields]
 
-        # Build a JSON decoder for intermediate data conversion (after delimited, before dataclass).
-        self._decoder: JSONDecoder[Any] = JSONDecoder(strict=False)
+        # Build a JSON decoder for intermediate data conversion (after delimited; before dataclass).
+        self._decoder: JSONDecoder[dict[str, JsonType]] = JSONDecoder(strict=False)
 
         # Build the delimited dictionary reader, filtering out any comment lines along the way.
         self._reader: DictReader[str] = DictReader(
             self._filter_out_comments(handle),
-            fieldnames=self._header if not header else None,
             delimiter=self.delimiter,
+            fieldnames=self._header if not header else None,
+            lineterminator=linesep,
             quotechar="'",
             quoting=csv.QUOTE_MINIMAL,
         )
 
         # Protect the user from the case where a header was specified, but a data line was found!
-        if self._reader.fieldnames is not None and (
-            set(self._reader.fieldnames) != set(self._header)
-        ):
+        if self._reader.fieldnames is not None and self._reader.fieldnames != self._header:
             raise ValueError("Fields of header do not match fields of dataclass!")
 
     @property
@@ -113,8 +114,8 @@
     def _filter_out_comments(self, lines: Iterator[str]) -> Iterator[str]:
         """Yield only lines in an iterator that do not start with a comment prefix."""
         for line in lines:
-            stripped: str = line.strip()
-            if not stripped:
+            self._line_count += 1
+            if not (stripped := line.strip()):
                 continue
             elif any(stripped.startswith(prefix) for prefix in self._comment_prefixes):
                 continue
@@ -126,7 +127,7 @@
         for record in self._reader:
             as_builtins = self._csv_dict_to_json(record)
             try:
-                yield convert(as_builtins, self._record_type, strict=False)
+                yield convert(as_builtins, self._record_type, strict=False, str_keys=True)
             except ValidationError as exception:
                 raise ValidationError(
                     "Could not parse JSON-like object into requested structure:"
@@ -152,14 +153,14 @@
             as_builtins: dict[str, JsonType] = self._decoder.decode(json_string)
         except DecodeError as exception:
             raise DecodeError(
-                "Could not load delimited data line into JSON-like format."
+                f"Could not load delimited data into JSON-like format on line {self._line_count}."
                 + f" Built improperly formatted JSON: {json_string}."
                 + f" Original exception: {exception}."
             ) from exception
 
         return as_builtins
 
     def _decode(self, field_type: type[Any] | str | Any, item: str) -> str:
         """A callback for overriding the string formatting of builtin and custom types."""
         if field_type is str:
             return f'"{item}"'

typeline/_writer.py

@@ -44,14 +44,14 @@
         self._handle: TextIOWrapper = handle
         self._record_type: type[RecordType] = record_type
 
-        # Inspect the record type and save the fields, field names, and field types.
+        # Inspect the record type and save the fields and field names.
         self._fields: tuple[Field[Any], ...] = fields_of(record_type)
-        self._header: list[str] = [field.name for field in fields_of(record_type)]
+        self._header: list[str] = [field.name for field in self._fields]
 
-        # Build a JSON encoder for intermediate data conversion (after dataclass, before delimited).
+        # Build a JSON encoder for intermediate data conversion (after dataclass; before delimited).
         self._encoder: JSONEncoder = JSONEncoder()
 
-        # Build the delimited dictionary reader, filtering out any comment lines along the way.
+        # Build the delimited dictionary writer which will use platform-dependent newlines.
         self._writer: DictWriter[str] = DictWriter(
             handle,
             fieldnames=self._header,
@@ -83,22 +83,19 @@
         self.close()
         return None
 
     def _encode(self, item: Any) -> Any:
-        """A callback for overriding the encoding of builtin types and custom types."""
-        if isinstance(item, tuple):
-            return list(item)  # pyright: ignore[reportUnknownVariableType, reportUnknownArgumentType]
+        """A custom encoder that can pre-process an item prior to serialization."""
         return item
 
     def write(self, record: RecordType) -> None:
         """Write the record to the open file-like object."""
         if not isinstance(record, self._record_type):
             raise ValueError(
-                f"Expected {self._record_type.__name__} but found"
-                + f" {record.__class__.__qualname__}!"
+                f"Expected {self._record_type.__name__} but found {record.__class__.__qualname__}!"
             )
 
         encoded = {name: self._encode(getattr(record, name)) for name in self._header}
         builtin = cast(dict[str, Any], to_builtins(encoded, str_keys=True))
         as_dict = {
             name: value if isinstance(value, str) else self._encoder.encode(value).decode("utf-8")
             for name, value in builtin.items()