v0.5.8 More control over header deduplication

Author: akariv

dataflows/VERSION

@@ -1 +1 @@
-0.5.7
+0.5.8

dataflows/processors/load.py

@@ -57,6 +57,8 @@ def __init__(self, load_source, name=None, resources=None, strip=True, limit_row
                  override_schema=None, override_fields=None,
                  extract_missing_values=None,
                  deduplicate_headers=False,
+                 deduplicate_headers_case_sensitive=True,
+                 deduplicate_headers_format=' (%s)',
                  on_error=raise_exception,
                  **options):
         super(load, self).__init__()
@@ -70,6 +72,8 @@ def __init__(self, load_source, name=None, resources=None, strip=True, limit_row
         self.override_schema = override_schema
         self.override_fields = override_fields
         self.deduplicate_headers = deduplicate_headers
+        self.deduplicate_headers_case_sensitive = deduplicate_headers_case_sensitive
+        self.deduplicate_headers_format = deduplicate_headers_format
 
         # Extract missing values
         self.extract_missing_values = None
@@ -180,12 +184,21 @@ def safe_process_datapackage(self, dp: Package):
                 self.options.setdefault('headers', 1)
                 self.options.setdefault('sample_size', 1000)
                 stream: Stream = Stream(self.load_source, **self.options).open()
-                if len(stream.headers) != len(set(stream.headers)):
+                if self.deduplicate_headers_case_sensitive:
+                    duplication_test = len(stream.headers) != len(set(stream.headers))
+                else:
+                    lower_headers = [header.lower() for header in stream.headers]
+                    duplication_test = len(lower_headers) != len(set(lower_headers))
+                # duplication_test = len(stream.headers) != len(set(stream.headers))
+                if duplication_test:
                     if not self.deduplicate_headers:
                         raise ValueError(
                             'Found duplicate headers.' +
                             'Use the `deduplicate_headers` flag (found headers=%r)' % stream.headers)
-                    stream.headers = self.rename_duplicate_headers(stream.headers)
+                    stream.headers = self.rename_duplicate_headers(
+                        stream.headers, case_sensitive=self.deduplicate_headers_case_sensitive,
+                        deduplicate_format=self.deduplicate_headers_format
+                    )
                 schema = Schema(self.override_schema or {}).infer(
                     stream.sample, headers=stream.headers,
                     confidence=1, guesser_cls=self.guesser)
@@ -269,15 +282,21 @@ def process_resources(self, resources):
             yield it
 
     @staticmethod
-    def rename_duplicate_headers(duplicate_headers):
+    def rename_duplicate_headers(duplicate_headers, case_sensitive=True, deduplicate_format=' (%s)'):
         counter = {}
         headers = []
+        header_keys = []
         for header in duplicate_headers:
-            counter.setdefault(header, 0)
-            counter[header] += 1
-            if counter[header] > 1:
-                if counter[header] == 2:
-                    headers[headers.index(header)] = '%s (%s)' % (header, 1)
-                header = '%s (%s)' % (header, counter[header])
+            header_key = header
+            header_keys.append(header_key)
+            if not case_sensitive:
+                header_key = header_key.lower()
+            counter.setdefault(header_key, 0)
+            counter[header_key] += 1
+            if counter[header_key] > 1:
+                if counter[header_key] == 2:
+                    prev_index = header_keys.index(header_key) 
+                    headers[prev_index] = ('%s' + deduplicate_format) % (headers[prev_index], 1)
+                header = ('%s' + deduplicate_format) % (header, counter[header_key])
             headers.append(header)
         return headers

tests/test_lib.py

@@ -1955,6 +1955,21 @@ def test_load_duplicate_headers():
     assert 'duplicate headers' in str(cause)
 
 
+def test_load_duplicate_headers_with_deduplicate_headers_flag_and_format():
+    from dataflows import load
+    flow = Flow(
+        load('data/duplicate_headers.csv', deduplicate_headers=True, deduplicate_headers_format='__%s'),
+    )
+    data, package, stats = flow.results()
+    assert package.descriptor['resources'][0]['schema']['fields'] == [
+        {'name': 'header1', 'type': 'string', 'format': 'default'},
+        {'name': 'header2__1', 'type': 'string', 'format': 'default'},
+        {'name': 'header2__2', 'type': 'string', 'format': 'default'},
+    ]
+    assert data == [[
+        {'header1': 'value1', 'header2__1': 'value2', 'header2__2': 'value3'},
+    ]]
+
 def test_load_duplicate_headers_with_deduplicate_headers_flag():
     from dataflows import load
     flow = Flow(
@@ -1970,6 +1985,40 @@ def test_load_duplicate_headers_with_deduplicate_headers_flag():
         {'header1': 'value1', 'header2 (1)': 'value2', 'header2 (2)': 'value3'},
     ]]
 
+def test_load_duplicate_headers_case():
+    from dataflows import load, exceptions
+    flow = Flow(
+        load('data/duplicate_headers_case.csv'),
+    )
+    data, package, stats = flow.results()
+    assert data == [[
+        {'header1': 'value1', 'header2': 'value2', 'HEADER2': 'value3'},
+    ]]
+
+    flow = Flow(
+        load('data/duplicate_headers_case.csv', deduplicate_headers_case_sensitive=False),
+    )
+    with pytest.raises(exceptions.ProcessorError) as excinfo:
+        flow.results()
+    cause = excinfo.value.cause
+    assert 'duplicate headers' in str(cause)
+
+
+def test_load_duplicate_headers_case_with_deduplicate_headers_flag():
+    from dataflows import load
+    flow = Flow(
+        load('data/duplicate_headers_case.csv', deduplicate_headers=True, deduplicate_headers_case_sensitive=False),
+    )
+    data, package, stats = flow.results()
+    assert package.descriptor['resources'][0]['schema']['fields'] == [
+        {'name': 'header1', 'type': 'string', 'format': 'default'},
+        {'name': 'header2 (1)', 'type': 'string', 'format': 'default'},
+        {'name': 'HEADER2 (2)', 'type': 'string', 'format': 'default'},
+    ]
+    assert data == [[
+        {'header1': 'value1', 'header2 (1)': 'value2', 'HEADER2 (2)': 'value3'},
+    ]]
+
 
 # Temporal format
 
@@ -2389,6 +2438,28 @@ def test_rename_fields_simple():
 
     assert res == [dict(A=i, B=i, c=i) for i in range(5)]
 
+
+def test_rename_fields_disable_regex():
+    from dataflows import Flow, rename_fields
+
+    data = []
+    for i in range(5):
+        x = dict()
+        x['a (1)'] = i
+        x['b (2)'] = i
+        x['c (3)'] = i
+        data.append(x)
+    rename = dict()
+    rename['a (1)'] = 'A'
+    rename['b (2)'] = 'B'
+    rename['c (3)'] = 'C'
+    res = Flow(
+        data,
+        rename_fields(rename, regex=False),
+    ).results()[0][0]
+
+    assert res == [dict(A=i, B=i, C=i) for i in range(5)]
+
 def test_rename_fields_regex():
     from dataflows import Flow, rename_fields