Repository: didactic-meme/pyfuncol (language: python)

Issue:
[FEATURE] Add parallel operations
Could be useful [this link](https://stackoverflow.com/questions/26520781/multiprocessing-pool-whats-the-difference-between-map-async-and-imap).


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index a44480e..7c1aca4 100644
--- a/README.md
+++ b/README.md
@@ -3,6 +3,7 @@
 ![CI](https://github.com/Gondolav/pyfuncol/actions/workflows/python-app.yml/badge.svg)
 [![codecov](https://codecov.io/gh/Gondolav/pyfuncol/branch/main/graph/badge.svg)](https://codecov.io/gh/Gondolav/pyfuncol)
 ![PyPI](https://img.shields.io/pypi/v/pyfuncol?color=blue)
+[![Downloads](https://pepy.tech/badge/pyfuncol/month)](https://pepy.tech/project/pyfuncol)
 [![Documentation Status](https://readthedocs.org/projects/pyfuncol/badge/?version=latest)](https://pyfuncol.readthedocs.io/en/latest/?badge=latest)
 [![GitHub license](https://img.shields.io/github/license/Gondolav/pyfuncol)](https://github.com/Gondolav/pyfuncol/blob/main/LICENSE)
 
@@ -31,7 +32,10 @@ import pyfuncol
 [1, 2, 3, 4].map(lambda x: x * 2).filter(lambda x: x > 4)
 # [6, 8]
 
-{1, 2, 3, 4}.map(lambda x: x * 2).filter(lambda x: x > 4)
+[1, 2, 3, 4].fold_left(0, lambda acc, n: acc + n)
+# 10
+
+{1, 2, 3, 4}.map(lambda x: x * 2).filter_not(lambda x: x <= 4)
 # {6, 8}
 
 ["abc", "def", "e"].group_by(lambda s: len(s))
@@ -39,6 +43,17 @@ import pyfuncol
 
 {"a": 1, "b": 2, "c": 3}.flat_map(lambda kv: {kv[0]: kv[1] ** 2})
 # {"a": 1, "b": 4, "c": 9}
+
+# pyfuncol also provides some parallel operations
+
+[1, 2, 3, 4].par_map(lambda x: x * 2).par_filter(lambda x: x > 4)
+# [6, 8]
+
+{1, 2, 3, 4}.par_map(lambda x: x * 2).par_filter(lambda x: x > 4)
+# {6, 8}
+
+{"a": 1, "b": 2, "c": 3}.par_flat_map(lambda kv: {kv[0]: kv[1] ** 2})
+# {"a": 1, "b": 4, "c": 9}
 ```
 
 ### API
diff --git a/docs/source/examples.md b/docs/source/examples.md
index a6d5a91..9ab8059 100644
--- a/docs/source/examples.md
+++ b/docs/source/examples.md
@@ -8,9 +8,28 @@ import pyfuncol
 [1, 2, 3, 4].map(lambda x: x * 2).filter(lambda x: x > 4)
 # [6, 8]
 
+[1, 2, 3, 4].fold_left(0, lambda acc, n: acc + n)
+# 10
+
+{1, 2, 3, 4}.map(lambda x: x * 2).filter_not(lambda x: x <= 4)
+# {6, 8}
+
 ["abc", "def", "e"].group_by(lambda s: len(s))
 # {3: ["abc", "def"], 1: ["e"]}
 
 {"a": 1, "b": 2, "c": 3}.flat_map(lambda kv: {kv[0]: kv[1] ** 2})
 # {"a": 1, "b": 4, "c": 9}
 ```
+
+pyfuncol also provides parallel operations (for now `par_map`, `par_flat_map`, `par_filter` and `par_filter_not`):
+
+```python
+[1, 2, 3, 4].par_map(lambda x: x * 2).par_filter(lambda x: x > 4)
+# [6, 8]
+
+{1, 2, 3, 4}.par_map(lambda x: x * 2).par_filter_not(lambda x: x <= 4)
+# {6, 8}
+
+{"a": 1, "b": 2, "c": 3}.par_flat_map(lambda kv: {kv[0]: kv[1] ** 2})
+# {"a": 1, "b": 4, "c": 9}
+```
diff --git a/pyfuncol/dict.py b/pyfuncol/dict.py
index 27bd5cc..6d02bbe 100644
--- a/pyfuncol/dict.py
+++ b/pyfuncol/dict.py
@@ -1,5 +1,6 @@
 from forbiddenfruit import curse
 from typing import Callable, Dict, Optional, Tuple, TypeVar, List
+import dask
 
 A = TypeVar("A")
 B = TypeVar("B")
@@ -230,6 +231,66 @@ def filter_not(self: Dict[A, B], p: Callable[[Tuple[A, B]], bool]) -> Dict[A, B]
     return {k: v for k, v in self.items() if not p((k, v))}
 
 
+# Parallel operations
+
+
+def par_filter(self: Dict[A, B], p: Callable[[Tuple[A, B]], bool]) -> Dict[A, B]:
+    """
+    Selects in parallel all elements of this dict which satisfy a predicate.
+
+    Args:
+        p: The predicate to satisfy.
+
+    Returns:
+        The filtered dict.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self.items()))
+    return {k: v for i, (k, v) in enumerate(self.items()) if preds[i]}
+
+
+def par_filter_not(self: Dict[A, B], p: Callable[[Tuple[A, B]], bool]) -> Dict[A, B]:
+    """
+    Selects in parallel all elements of this dict which do not satisfy a predicate.
+
+    Args:
+        p: The predicate to not satisfy.
+
+    Returns:
+        The filtered dict.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self.items()))
+    return {k: v for i, (k, v) in enumerate(self.items()) if not preds[i]}
+
+
+def par_flat_map(
+    self: Dict[A, B], f: Callable[[Tuple[A, B]], Dict[C, D]]
+) -> Dict[C, D]:
+    """
+    Builds a new dict by applying a function in parallel to all elements of this dict and using the elements of the resulting collections.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new dict.
+    """
+    applications = dask.compute(*(dask.delayed(f)(x) for x in self.items()))
+    return {k: v for y in applications for k, v in y.items()}
+
+
+def par_map(self: Dict[A, B], f: Callable[[Tuple[A, B]], Tuple[C, D]]) -> Dict[C, D]:
+    """
+    Builds a new dict by applying a function in parallel to all elements of this dict.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new dict.
+    """
+    return dict(dask.compute(*(dask.delayed(f)(x) for x in self.items())))
+
+
 def extend_dict():
     """
     Extends the dict built-in type with methods.
@@ -248,3 +309,9 @@ def extend_dict():
     curse(dict, "forall", forall)
     curse(dict, "find", find)
     curse(dict, "filter_not", filter_not)
+
+    # Parallel operations
+    curse(dict, "par_map", par_map)
+    curse(dict, "par_filter", par_filter)
+    curse(dict, "par_filter_not", par_filter_not)
+    curse(dict, "par_flat_map", par_flat_map)
diff --git a/pyfuncol/list.py b/pyfuncol/list.py
index e05fe2c..f2430b2 100644
--- a/pyfuncol/list.py
+++ b/pyfuncol/list.py
@@ -1,6 +1,7 @@
 from forbiddenfruit import curse
 from collections import defaultdict
 from typing import Callable, Dict, Optional, TypeVar, List
+import dask
 
 A = TypeVar("A")
 B = TypeVar("B")
@@ -270,6 +271,65 @@ def length(self: List[A]) -> int:
     return len(self)
 
 
+# Parallel operations
+
+
+def par_map(self: List[A], f: Callable[[A], B]) -> List[B]:
+    """
+    Builds a new list by applying a function in parallel to all elements of this list.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new list.
+    """
+    return list(dask.compute(*(dask.delayed(f)(x) for x in self)))
+
+
+def par_filter(self: List[A], p: Callable[[A], bool]) -> List[A]:
+    """
+    Selects in parallel all elements of this list which satisfy a predicate.
+
+    Args:
+        p: The predicate to satisfy.
+
+    Returns:
+        The filtered list.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self))
+    return [x for i, x in enumerate(self) if preds[i]]
+
+
+def par_filter_not(self: List[A], p: Callable[[A], bool]) -> List[A]:
+    """
+    Selects in parallel all elements of this list which do not satisfy a predicate.
+
+    Args:
+        p: The predicate to not satisfy.
+
+    Returns:
+        The filtered list.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self))
+    return [x for i, x in enumerate(self) if not preds[i]]
+
+
+def par_flat_map(self: List[A], f: Callable[[A], List[B]]) -> List[B]:
+    """
+    Builds a new list by applying a function in parallel to all elements of this list and using the
+    elements of the resulting collections.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new list.
+    """
+    applications = dask.compute(*(dask.delayed(f)(x) for x in self))
+    return [z for y in applications for z in y]
+
+
 def extend_list():
     """
     Extends the list built-in type with methods.
@@ -293,3 +353,9 @@ def extend_list():
     curse(list, "tail", tail)
     curse(list, "take", take)
     curse(list, "length", length)
+
+    # Parallel operations
+    curse(list, "par_map", par_map)
+    curse(list, "par_filter", par_filter)
+    curse(list, "par_filter_not", par_filter_not)
+    curse(list, "par_flat_map", par_flat_map)
diff --git a/pyfuncol/set.py b/pyfuncol/set.py
index 740086f..a11659b 100644
--- a/pyfuncol/set.py
+++ b/pyfuncol/set.py
@@ -1,6 +1,7 @@
 from forbiddenfruit import curse
 from collections import defaultdict
 from typing import Callable, Dict, Optional, TypeVar, Set
+import dask
 
 A = TypeVar("A")
 B = TypeVar("B")
@@ -193,6 +194,64 @@ def length(self: Set[A]) -> int:
     return len(self)
 
 
+# Parallel operations
+
+
+def par_map(self: Set[A], f: Callable[[A], B]) -> Set[B]:
+    """
+    Builds a new set by applying in parallel a function to all elements of this set.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new set.
+    """
+    return set(dask.compute(*(dask.delayed(f)(x) for x in self)))
+
+
+def par_filter(self: Set[A], p: Callable[[A], bool]) -> Set[A]:
+    """
+    Selects in parallel all elements of this set which satisfy a predicate.
+
+    Args:
+        p: The predicate to satisfy.
+
+    Returns:
+        The filtered set.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self))
+    return {x for i, x in enumerate(self) if preds[i]}
+
+
+def par_filter_not(self: Set[A], p: Callable[[A], bool]) -> Set[A]:
+    """
+    Selects in parallel all elements of this set which do not satisfy a predicate.
+
+    Args:
+        p: The predicate to not satisfy.
+
+    Returns:
+        The filtered set.
+    """
+    preds = dask.compute(*(dask.delayed(p)(x) for x in self))
+    return {x for i, x in enumerate(self) if not preds[i]}
+
+
+def par_flat_map(self: Set[A], f: Callable[[A], Set[B]]) -> Set[B]:
+    """
+    Builds a new set by applying in parallel a function to all elements of this set and using the elements of the resulting collections.
+
+    Args:
+        f: The function to apply to all elements.
+
+    Returns:
+        The new set.
+    """
+    applications = dask.compute(*(dask.delayed(f)(x) for x in self))
+    return {x for y in applications for x in y}
+
+
 def extend_set():
     """
     Extends the set built-in type with methods.
@@ -210,3 +269,9 @@ def extend_set():
     curse(set, "fold_right", fold_right)
     curse(set, "forall", forall)
     curse(set, "length", length)
+
+    # Parallel operations
+    curse(set, "par_map", par_map)
+    curse(set, "par_filter", par_filter)
+    curse(set, "par_filter_not", par_filter_not)
+    curse(set, "par_flat_map", par_flat_map)
diff --git a/requirements.txt b/requirements.txt
index 799d689..4d671c3 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,1 +1,2 @@
+dask==2021.12.0
 forbiddenfruit==0.1.4
\ No newline at end of file


Test patch:
diff --git a/pyfuncol/tests/test_dict.py b/pyfuncol/tests/test_dict.py
index 2eadec1..14bf2c6 100644
--- a/pyfuncol/tests/test_dict.py
+++ b/pyfuncol/tests/test_dict.py
@@ -69,3 +69,22 @@ def test_find_none():
 
 def test_filter_not():
     assert d.filter_not(lambda kv: kv[1] > 1) == {"a": 1}
+
+
+# Parallel operations
+
+
+def test_par_filter():
+    assert d.par_filter(lambda kv: kv[1] > 1) == {"b": 2, "c": 3}
+
+
+def test_par_filter_not():
+    assert d.par_filter_not(lambda kv: kv[1] <= 1) == {"b": 2, "c": 3}
+
+
+def test_par_flat_map():
+    assert d.par_flat_map(lambda kv: {kv[0]: kv[1] ** 2}) == {"a": 1, "b": 4, "c": 9}
+
+
+def test_par_map():
+    assert d.par_map(lambda kv: (kv[0], kv[1] ** 2)) == {"a": 1, "b": 4, "c": 9}
diff --git a/pyfuncol/tests/test_list.py b/pyfuncol/tests/test_list.py
index e04d233..34cc649 100644
--- a/pyfuncol/tests/test_list.py
+++ b/pyfuncol/tests/test_list.py
@@ -134,3 +134,22 @@ def test_length():
 
 def test_length_equal_size():
     assert l.size() == l.length()
+
+
+# Parallel operations
+
+
+def test_par_map():
+    assert l.par_map(lambda x: x * 2) == [2, 4, 6]
+
+
+def test_par_filter():
+    assert l.par_filter(lambda x: x >= 2) == [2, 3]
+
+
+def test_par_filter_not():
+    assert l.par_filter_not(lambda x: x < 2) == [2, 3]
+
+
+def test_par_flat_map():
+    assert l.par_flat_map(lambda x: [x ** 2]) == [1, 4, 9]
diff --git a/pyfuncol/tests/test_set.py b/pyfuncol/tests/test_set.py
index aa1840e..ae27860 100644
--- a/pyfuncol/tests/test_set.py
+++ b/pyfuncol/tests/test_set.py
@@ -1,5 +1,4 @@
 import pyfuncol
-import pytest
 
 s = {1, 2, 3}
 
@@ -89,3 +88,22 @@ def test_length():
 
 def test_length_equal_size():
     assert s.size() == s.length()
+
+
+# Parallel operations
+
+
+def test_par_map():
+    assert s.par_map(lambda x: x * 2) == {2, 4, 6}
+
+
+def test_par_filter():
+    assert s.par_filter(lambda x: x >= 2) == {2, 3}
+
+
+def test_par_filter_not():
+    assert s.par_filter_not(lambda x: x < 2) == {2, 3}
+
+
+def test_par_flat_map():
+    assert s.par_flat_map(lambda x: [x ** 2]) == {1, 4, 9}