- Module multiprocessing

- Method `dict_utils.dict_intersection`

CHANGELOG.md

@@ -4,6 +4,13 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [0.7.2] - 23.04.2023
+
+### Added
+
+- Module `multiprocessing`
+- Method `dict_utils.dict_intersection`
+
 ## [0.7.1] - 23.04.2023
 
 ### Added

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = DRSlib-DavidRodriguezSoaresCUI
-version = 0.7.1
+version = 0.7.2
 author = DavidRodriguezSoaresCUI
 author_email = fireblaze904+DRSlib@gmail.com
 description = DRSlib - a set of utilities by DavidRodriguezSoaresCUI

src/DRSlib/dict_utils.py

@@ -54,6 +54,38 @@ def dict_difference(dictA: dict, dictB: dict) -> dict:
     return diff
 
 
+def dict_intersection(dicts: List[dict]) -> dict:
+    """Given a list of dictionnaries, returns the common elements
+    determined by key
+    """
+    assertTrue(
+        len(dicts) > 1, "Expected at least 2 dictionnaries, found {}", len(dicts)
+    )
+    assertTrue(
+        all(d is not None and isinstance(d, dict) for d in dicts),
+        "Invalid argument: some items are Nore or not dicts!",
+    )
+
+    common = {}
+
+    for k, vref in dicts[0].items():
+        if not all(k in d for d in dicts[1:]):
+            # Some dicts don't have key `k`
+            continue
+        if isinstance(vref, dict):
+            common_v = dict_intersection([d[k] for d in dicts])
+            common[k] = common_v if common_v else "<varies>"
+            continue
+        value_set = set(d[k] for d in dicts)
+        if len(value_set) != 1:
+            # Divergent values for key `k`
+            common[k] = "<varies>"
+            continue
+        common[k] = vref
+
+    return common
+
+
 def dict_list_keys(d: dict) -> List[str]:
     """Searches (recursively) for all keys within dictionnary and returns them in an ordered list.
     Warns on duplicate."""

src/DRSlib/multiprocessing.py

@@ -0,0 +1,39 @@
+"""
+Multiprocessing
+===============
+
+For when even the most basic parallel processing is
+better than nothing
+"""
+import multiprocessing
+from typing import Any, Callable, List, Tuple
+
+
+class SimpleMultiProcessing:
+    """Parallel processing made simpler"""
+
+    @staticmethod
+    def apply_kwargs(
+        user_function_and_arguments: Tuple[Callable, tuple, dict]
+    ) -> Callable:
+        """Workaround for imap only taking one argument per thread
+        Executes user function with provided arguments"""
+        user_function, args, kwargs = user_function_and_arguments
+        return user_function(*args, **kwargs)
+
+    @staticmethod
+    def bulk_processing(
+        user_function: Callable, arguments: List[dict], parallel_instances: int
+    ) -> List[Any]:
+        """Takes in callable, kwargs arguments for each thread and number of instances
+        to execute in parallel at a time. Returns execution's return value in order.
+        """
+        user_function_and_arguments = [
+            (user_function, (), _args) for _args in arguments
+        ]
+        with multiprocessing.Pool(parallel_instances) as pool:
+            return list(
+                pool.imap(
+                    SimpleMultiProcessing.apply_kwargs, user_function_and_arguments
+                )
+            )