checkpoint

MANIFEST.in

@@ -1 +0,0 @@
-include VERSION

README.rst

@@ -2,17 +2,28 @@
 deepmerge
 =========
 
+.. image:: https://img.shields.io/pypi/v/deepmerge.svg
+   :target: https://pypi.org/project/deepmerge/
+
+.. image:: https://img.shields.io/pypi/status/deepmerge.svg
+   :target: https://pypi.org/project/deepmerge/
+
+.. image:: https://img.shields.io/pypi/pyversions/pillar.svg
+   :target: https://github.com/toumorokoshi/deepmerge
+
+.. image:: https://img.shields.io/github/license/toumorokoshi/deepmerge.svg
+   :target: https://github.com/toumorokoshi/deepmerge
+
 .. image:: https://github.com/toumorokoshi/deepmerge/actions/workflows/python-package.yaml/badge.svg
     :target: https://github.com/toumorokoshi/deepmerge/actions/workflows/python-package.yaml
 
-A tools to handle merging of
-nested data structures in python.
+A tool to handle merging of nested data structures in Python.
 
 ------------
 Installation
 ------------
 
-deepmerge is available on `pypi <https://pypi.python.org/>`_:
+deepmerge is available on `pypi <https://pypi.org/project/deepmerge/>`_:
 
 .. code-block:: bash
 
@@ -67,4 +78,4 @@ Example
 
 You can also pass in your own merge functions, instead of a string.
 
-For more information, see the `docs <https://deepmerge.readthedocs.io/en/latest/>`_
+For more information, see the `docs <https://deepmerge.readthedocs.io/en/latest/>`_

deepmerge/__init__.py

@@ -1,9 +1,11 @@
+from typing import List, Tuple, Type
+
 from .merger import Merger
 from .strategy.core import STRATEGY_END  # noqa
 
 # some standard mergers available
 
-DEFAULT_TYPE_SPECIFIC_MERGE_STRATEGIES = [
+DEFAULT_TYPE_SPECIFIC_MERGE_STRATEGIES: List[Tuple[Type, str]] = [
     (list, "append"),
     (dict, "merge"),
     (set, "union"),

deepmerge/exception.py

@@ -1,18 +1,35 @@
+from __future__ import annotations
+
+from typing import List, Any
+
+import deepmerge.merger
+
+
 class DeepMergeException(Exception):
-    pass
+    "Base class for all `deepmerge` Exceptions"
 
 
 class StrategyNotFound(DeepMergeException):
-    pass
+    "Exception for when a strategy cannot be located"
 
 
 class InvalidMerge(DeepMergeException):
-    def __init__(self, strategy_list_name, merge_args, merge_kwargs):
-        super(InvalidMerge, self).__init__(
-            "no more strategies found for {0} and arguments {1}, {2}".format(
-                strategy_list_name, merge_args, merge_kwargs
-            )
+    "Exception for when unable to complete a merge operation"
+
+    def __init__(
+        self,
+        strategy_list_name: str,
+        config: deepmerge.merger.Merger,
+        path: List,
+        base: Any,
+        nxt: Any,
+    ) -> None:
+        super().__init__(
+            f"Could not merge using {strategy_list_name!r} [{config=}, {path=}, {base=}, {nxt=}]"
         )
         self.strategy_list_name = strategy_list_name
-        self.merge_args = merge_args
-        self.merge_kwargs = merge_kwargs
+        self.config = config
+        self.path = path
+        self.base = base
+        self.nxt = nxt
+        return

deepmerge/extended_set.py

@@ -1,3 +1,6 @@
+from typing import Sequence, Any
+
+
 class ExtendedSet(set):
     """
     ExtendedSet is an extension of set, which allows for usage
@@ -9,17 +12,17 @@ class ExtendedSet(set):
     - unhashable types
     """
 
-    def __init__(self, elements):
-        self._values_by_hash = {self._hash(e): e for e in elements}
+    def __init__(self, elements: Sequence) -> None:
+        self._values_by_hash = {self._hash_element(e): e for e in elements}
 
-    def _insert(self, element):
-        self._values_by_hash[self._hash(element)] = element
+    def _insert(self, element: Any) -> None:
+        self._values_by_hash[self._hash_element(element)] = element
+        return
 
-    def _hash(self, element):
+    def _hash_element(self, element: Any) -> int:
         if getattr(element, "__hash__") is not None:
             return hash(element)
-        else:
-            return hash(str(element))
+        return hash(str(element))
 
-    def __contains__(self, obj):
-        return self._hash(obj) in self._values_by_hash
+    def __contains__(self, obj: Any) -> bool:
+        return self._hash_element(obj) in self._values_by_hash

deepmerge/merger.py

@@ -1,44 +1,66 @@
-from .strategy.list import ListStrategies
-from .strategy.dict import DictStrategies
-from .strategy.set import SetStrategies
-from .strategy.type_conflict import TypeConflictStrategies
-from .strategy.fallback import FallbackStrategies
+from __future__ import annotations
 
+from typing import Dict, List, Type, Tuple, Any, Union, Sequence, Callable
 
-class Merger(object):
+from . import strategy as s
+
+
+class Merger:
     """
+    Merges objects based on provided strategies
+
     :param type_strategies, List[Tuple]: a list of (Type, Strategy) pairs
            that should be used against incoming types. For example: (dict, "override").
     """
 
-    PROVIDED_TYPE_STRATEGIES = {
-        list: ListStrategies,
-        dict: DictStrategies,
-        set: SetStrategies,
+    PROVIDED_TYPE_STRATEGIES: Dict[Type, Type[s.StrategyList]] = {
+        list: s.ListStrategies,
+        dict: s.DictStrategies,
+        set: s.SetStrategies,
     }
 
-    def __init__(self, type_strategies, fallback_strategies, type_conflict_strategies):
-        self._fallback_strategy = FallbackStrategies(fallback_strategies)
+    def __init__(
+        self,
+        type_strategies: Sequence[
+            Tuple[Type, Union[s.StrategyCallable, s.StrategyListInitable]]
+        ],
+        fallback_strategies: s.StrategyListInitable,
+        type_conflict_strategies: s.StrategyListInitable,
+    ) -> None:
+        self._fallback_strategy = s.FallbackStrategies(fallback_strategies)
+        self._type_conflict_strategy = s.TypeConflictStrategies(
+            type_conflict_strategies
+        )
 
-        expanded_type_strategies = []
+        self._type_strategies: List[Tuple[Type, s.StrategyCallable]] = []
         for typ, strategy in type_strategies:
             if typ in self.PROVIDED_TYPE_STRATEGIES:
-                strategy = self.PROVIDED_TYPE_STRATEGIES[typ](strategy)
-            expanded_type_strategies.append((typ, strategy))
-        self._type_strategies = expanded_type_strategies
+                # Customise a StrategyList instance for this type
+                self._type_strategies.append(
+                    (typ, self.PROVIDED_TYPE_STRATEGIES[typ](strategy))
+                )
+            elif callable(strategy):
+                self._type_strategies.append((typ, strategy))
+            else:
+                raise ValueError(f"Cannot handle ({typ}, {strategy})")
+        return
 
-        self._type_conflict_strategy = TypeConflictStrategies(type_conflict_strategies)
-
-    def merge(self, base, nxt):
+    def merge(self, base: Any, nxt: Any) -> Any:
         return self.value_strategy([], base, nxt)
 
-    def type_conflict_strategy(self, *args):
-        return self._type_conflict_strategy(self, *args)
+    def type_conflict_strategy(self, path: List, base: Any, nxt: Any) -> Any:
+        return self._type_conflict_strategy(self, path, base, nxt)
 
-    def value_strategy(self, path, base, nxt):
+    def value_strategy(self, path: List, base: Any, nxt: Any) -> Any:
+        # Check for strategy based on type of base, next
         for typ, strategy in self._type_strategies:
             if isinstance(base, typ) and isinstance(nxt, typ):
+                # We have a strategy for this type
                 return strategy(self, path, base, nxt)
-        if not (isinstance(base, type(nxt)) or isinstance(nxt, type(base))):
-            return self.type_conflict_strategy(path, base, nxt)
-        return self._fallback_strategy(self, path, base, nxt)
+
+        if isinstance(base, type(nxt)) or isinstance(nxt, type(base)):
+            # no known strategy but base, next are similar types
+            return self._fallback_strategy(self, path, base, nxt)
+
+        # No known strategy and base, next are different types.
+        return self.type_conflict_strategy(path, base, nxt)

deepmerge/strategy/__init__.py

@@ -0,0 +1,6 @@
+from .core import StrategyList, StrategyCallable, StrategyListInitable
+from .dict import DictStrategies
+from .fallback import FallbackStrategies
+from .list import ListStrategies
+from .set import SetStrategies
+from .type_conflict import TypeConflictStrategies

deepmerge/strategy/core.py

@@ -1,20 +1,38 @@
+from __future__ import annotations
+
+import sys
+from typing import TYPE_CHECKING, List, Callable, Any, Union
+
+if sys.version_info >= (3, 10):
+    from typing import TypeAlias
+else:
+    from typing_extentions import TypeAlias
+
+import deepmerge.merger
 from ..exception import StrategyNotFound, InvalidMerge
-from ..compat import string_type
 
 STRATEGY_END = object()
 
+StrategyCallable: TypeAlias = "Callable[[deepmerge.merger.Merger, List, Any, Any], Any]"
+StrategyListInitable: TypeAlias = (
+    "Union[str, StrategyCallable, List[Union[str, StrategyCallable]]]"
+)
 
-class StrategyList(object):
 
-    NAME = None
+class StrategyList:
+    NAME: str
 
-    def __init__(self, strategy_list):
+    def __init__(self, strategy_list: StrategyListInitable) -> None:
         if not isinstance(strategy_list, list):
             strategy_list = [strategy_list]
-        self._strategies = [self._expand_strategy(s) for s in strategy_list]
+        self._strategies: List[StrategyCallable] = [
+            self._expand_strategy(s) for s in strategy_list
+        ]
 
     @classmethod
-    def _expand_strategy(cls, strategy):
+    def _expand_strategy(
+        cls, strategy: Union[str, StrategyCallable]
+    ) -> StrategyCallable:
         """
         :param strategy: string or function
 
@@ -23,16 +41,18 @@ def _expand_strategy(cls, strategy):
 
         Otherwise, return the value, implicitly assuming it's a function.
         """
-        if isinstance(strategy, string_type):
-            method_name = "strategy_{0}".format(strategy)
-            if not hasattr(cls, method_name):
-                raise StrategyNotFound(strategy)
-            return getattr(cls, method_name)
+        if isinstance(strategy, str):
+            method_name = f"strategy_{strategy}"
+            if hasattr(cls, method_name):
+                return getattr(cls, method_name)
+            raise StrategyNotFound(strategy)
         return strategy
 
-    def __call__(self, *args, **kwargs):
+    def __call__(
+        self, config: deepmerge.merger.Merger, path: List, base: Any, nxt: Any
+    ) -> Any:
         for s in self._strategies:
-            ret_val = s(*args, **kwargs)
+            ret_val = s(config, path, base, nxt)
             if ret_val is not STRATEGY_END:
                 return ret_val
-        raise InvalidMerge(self.NAME, args, kwargs)
+        raise InvalidMerge(self.NAME, config, path, base, nxt)

deepmerge/strategy/dict.py

@@ -1,3 +1,8 @@
+from __future__ import annotations
+
+from typing import Dict, List
+
+import deepmerge.merger
 from .core import StrategyList
 
 
@@ -10,7 +15,9 @@ class DictStrategies(StrategyList):
     NAME = "dict"
 
     @staticmethod
-    def strategy_merge(config, path, base, nxt):
+    def strategy_merge(
+        config: deepmerge.merger.Merger, path: List, base: Dict, nxt: Dict
+    ) -> Dict:
         """
         for keys that do not exists,
         use them directly. if the key exists
@@ -24,7 +31,9 @@ def strategy_merge(config, path, base, nxt):
         return base
 
     @staticmethod
-    def strategy_override(config, path, base, nxt):
+    def strategy_override(
+        config: deepmerge.merger.Merger, path: List, base: Dict, nxt: Dict
+    ) -> Dict:
         """
         move all keys in nxt into base, overriding
         conflicts.

deepmerge/strategy/fallback.py

@@ -1,6 +1,14 @@
+from __future__ import annotations
+
+from typing import List, Any, TypeVar
+
+import deepmerge.merger
 from .core import StrategyList
 
 
+T = TypeVar("T")
+
+
 class FallbackStrategies(StrategyList):
     """
     The StrategyList containing fallback strategies.
@@ -9,11 +17,15 @@ class FallbackStrategies(StrategyList):
     NAME = "fallback"
 
     @staticmethod
-    def strategy_override(config, path, base, nxt):
+    def strategy_override(
+        config: deepmerge.merger.Merger, path: List, base: Any, nxt: T
+    ) -> T:
         """use nxt, and ignore base."""
         return nxt
 
     @staticmethod
-    def strategy_use_existing(config, path, base, nxt):
+    def strategy_use_existing(
+        config: deepmerge.merger.Merger, path: List, base: T, nxt: Any
+    ) -> T:
         """use base, and ignore next."""
         return base