v0.0.65 Introduce deduplicate processor

Author: akariv

PROCESSORS.md

@@ -567,6 +567,23 @@ def filter_rows(condition=None, equals=tuple(), not_equals=tuple(), resources=No
   - `None` indicates operation should be done on all resources
   - The index of the resource in the package
 
+#### deduplicate.py
+Deduplicates rows in resources based on the resources' primary key
+
+`deduplicate` accepts a resource specifier.
+For each resource, it will output only unique rows (based on the values in the primary key fields). Rows with duplicate primary keys will be ignored.
+
+```python
+def deduplicate(resources=None):
+    pass
+```
+
+- `resources`
+  - A name of a resource to operate on
+  - A regular expression matching resource names
+  - A list of resource names
+  - `None` indicates operation should be done on all resources
+  - The index of the resource in the package
 
 ### Manipulate package
 #### update_package.py

dataflows/VERSION

@@ -1 +1 @@
-0.0.64
+0.0.65

dataflows/helpers/iterable_loader.py

@@ -88,7 +88,7 @@ def handle_iterable(self):
         try:
             for x in self.iterable:
                 if mode is None:
-                    assert isinstance(x, (dict, list)), 'Bad item %r' % x
+                    assert isinstance(x, (dict, list, tuple)), 'Bad item %r' % x
                     mode = dict if isinstance(x, dict) else list
                 assert isinstance(x, mode)
                 if mode == dict:

dataflows/processors/__init__.py

@@ -9,6 +9,7 @@
 from .checkpoint import checkpoint
 from .concatenate import concatenate
 from .delete_fields import delete_fields
+from .deduplicate import deduplicate
 from .duplicate import duplicate
 from .filter_rows import filter_rows
 from .find_replace import find_replace

dataflows/processors/deduplicate.py

@@ -0,0 +1,31 @@
+from dataflows import PackageWrapper, ResourceWrapper
+
+from ..helpers.resource_matcher import ResourceMatcher
+
+
+def deduper(rows: ResourceWrapper):
+    pk = rows.res.descriptor['schema'].get('primaryKey', [])
+    if len(pk) == 0:
+        yield from rows
+    else:
+        keys = set()
+        for row in rows:
+            key = tuple(row[k] for k in pk)
+            if key in keys:
+                continue
+            keys.add(key)
+            yield row
+
+
+def deduplicate(resources=None):
+
+    def func(package: PackageWrapper):
+        resource_matcher = ResourceMatcher(resources, package)
+        yield package.pkg
+        resource: ResourceWrapper
+        for resource in package:
+            if resource_matcher.match(resource.res.name):
+                yield deduper(resource)
+            else:
+                yield resource
+    return func

tests/test_lib.py

@@ -471,6 +471,26 @@ def test_sort_rows_datetime():
     ]
 
 
+def test_deduplicate():
+    from dataflows import deduplicate, set_primary_key
+
+    a = [
+            {'a': 1, 'b': 3, 'c': 'First'},
+            {'a': 2, 'b': 3, 'c': 'First'},
+            {'a': 1, 'b': 3, 'c': '!First'},
+            {'a': 1, 'b': 2, 'c': 'First'},
+            {'a': 2, 'b': 3, 'c': '!First'},
+        ]
+
+    f = Flow(
+        a,
+        set_primary_key(['a', 'b']),
+        deduplicate(),
+    )
+    results, _, _ = f.results()
+    assert set(x['c'] for x in results[0]) == {'First'}
+
+
 def test_duplicate():
     from dataflows import duplicate