Validate duplicated header (dfurtado#42)

* Duplicate header validation

* Added header validation tests

* Updated README

Author: dfurtado

README.md

@@ -85,6 +85,16 @@ dataclass_csv.DataclassReader(
 )
 ```
 
+All keyword arguments support by `DictReader` are supported by the `DataclassReader`, with the addition of:
+
+`validate_header` - The `DataclassReader` will raise a `ValueError` if the CSV file cointain columns with the same name. This
+validation is performed to avoid data being overwritten. To skip this validation set `validate_header=False` when creating a
+instance of the `DataclassReader`, see an example below:
+
+```python
+reader = DataclassReader(f, User, validate_header=False)
+```
+
 If you run this code you should see an output like this:
 
 ```python

dataclass_csv/dataclass_reader.py

@@ -8,6 +8,26 @@
 from .field_mapper import FieldMapper
 from .exceptions import CsvValueError
 
+from collections import Counter
+
+
+def _verify_duplicate_header_items(header):
+    if header is not None and len(header) == 0:
+        return
+
+    header_counter = Counter(header)
+    duplicated = [k for k, v in header_counter.items() if v > 1]
+
+    if len(duplicated) > 0:
+        raise ValueError(
+            (
+                "It seems like the CSV file contain duplicated header "
+                f"values: {duplicated}. This may cause inconsistent data. "
+                "Use the kwarg validate_header=False when initializing the "
+                "DataclassReader to skip the header validation."
+            )
+        )
+
 
 class DataclassReader:
     def __init__(
@@ -32,10 +52,15 @@ def __init__(
         self._optional_fields = self._get_optional_fields()
         self._field_mapping: Dict[str, Dict[str, Any]] = {}
 
+        validate_header = kwds.pop("validate_header", True)
+
         self._reader = csv.DictReader(
             f, fieldnames, restkey, restval, dialect, *args, **kwds
         )
 
+        if validate_header:
+            _verify_duplicate_header_items(self._reader.fieldnames)
+
     def _get_optional_fields(self):
         return [
             field.name

tests/conftest.py

@@ -5,20 +5,20 @@
 
 @pytest.fixture()
 def create_csv(tmpdir_factory):
-    def func(data, filename='user.csv', factory=tmpdir_factory):
+    def func(data, fieldnames=None, filename="user.csv", factory=tmpdir_factory):
 
         assert data
 
-        file = tmpdir_factory.mktemp('data').join(filename)
+        file = tmpdir_factory.mktemp("data").join(filename)
 
         row = data[0] if isinstance(data, list) else data
 
-        with file.open('w') as f:
-            writer = DictWriter(f, fieldnames=row.keys())
+        header = fieldnames if fieldnames is not None else row.keys()
+
+        with file.open("w") as f:
+            writer = DictWriter(f, fieldnames=header)
             writer.writeheader()
-            addrow = (
-                writer.writerows if isinstance(data, list) else writer.writerow
-            )
+            addrow = writer.writerows if isinstance(data, list) else writer.writerow
             addrow(data)
 
         return file

tests/test_dataclass_reader.py

@@ -225,3 +225,26 @@ def test_raise_error_when_field_not_found(create_csv):
         ):
             reader = DataclassReader(f, UserWithEmail)
             list(reader)
+
+
+def test_raise_error_when_duplicate_header_items(create_csv):
+    csv_file = create_csv(
+        {"name": "User1", "email": "[email protected]"},
+        fieldnames=["name", "email", "name"],
+    )
+
+    with csv_file.open() as f:
+        with pytest.raises(ValueError):
+            reader = DataclassReader(f, UserWithEmail)
+            list(reader)
+
+
+def test_skip_header_validation(create_csv):
+    csv_file = create_csv(
+        {"name": "User1", "email": "[email protected]"},
+        fieldnames=["name", "email", "name"],
+    )
+
+    with csv_file.open() as f:
+        reader = DataclassReader(f, UserWithEmail, validate_header=False)
+        list(reader)