Mercurial > hgrepos > Python > libs > ConfigMix
diff configmix/__init__.py @ 741:e069797f0e36
Implemented the new merge stragegies when merging lists: "extend" and "prepend"
| author | Franz Glasner <fzglas.hg@dom66.de> |
|---|---|
| date | Sun, 29 Oct 2023 17:15:41 +0100 |
| parents | 324ae9a56a75 |
| children | e4fad9cdd906 |
line wrap: on
line diff
--- a/configmix/__init__.py Sun Oct 29 17:13:32 2023 +0100 +++ b/configmix/__init__.py Sun Oct 29 17:15:41 2023 +0100 @@ -57,6 +57,11 @@ :keyword strict: enable strict parsing mode for parsers that support it (e.g. to prevent duplicate keys) :type strict: bool + :keyword merge_lists: When ``None`` then lists will be overwritten + by the merge process. When ``extend`` then + lists will be extended instead. + This parameter is passed to :func:`.merge`. + :type merge_lists: str or None :returns: the configuration :rtype: ~configmix.config.Configuration @@ -64,22 +69,23 @@ defaults = kwargs.get("defaults") extras = kwargs.get("extras") strict = kwargs.get("strict", False) + merge_lists = kwargs.get("merge_lists", None) if defaults is None: ex = Configuration() else: - ex = merge(None, Configuration(defaults)) + ex = merge(None, Configuration(defaults), merge_lists=merge_lists) for f in files: if f.startswith(constants.DIR_PREFIX): for f2 in _get_configuration_files_from_dir(f[5:]): nx = _load_cfg_from_file(f2, ignore_unknown=True, strict=strict) if nx is not None: - ex = merge(nx, ex) + ex = merge(nx, ex, merge_lists=merge_lists) else: nx = _load_cfg_from_file(f, strict=strict) if nx is not None: - ex = merge(nx, ex) + ex = merge(nx, ex, merge_lists=merge_lists) if extras: - ex = merge(Configuration(extras), ex) + ex = merge(Configuration(extras), ex, merge_lists=merge_lists) return Configuration(ex) @@ -91,22 +97,23 @@ defaults = kwargs.get("defaults") extras = kwargs.get("extras") strict = kwargs.get("strict", False) + merge_lists = kwargs.get("merge_lists", None) if defaults is None: ex = Configuration() else: - ex = safe_merge(None, Configuration(defaults)) + ex = safe_merge(None, Configuration(defaults), merge_lists=merge_lists) for f in files: if f.startswith(constants.DIR_PREFIX): for f2 in _get_configuration_files_from_dir(f[5:]): nx = _load_cfg_from_file(f2, ignore_unknown=True, strict=strict) if nx is not None: - ex = safe_merge(nx, ex) + ex = safe_merge(nx, ex, merge_lists=merge_lists) else: nx = _load_cfg_from_file(f, strict=strict) if nx is not None: - ex = safe_merge(nx, ex) + ex = safe_merge(nx, ex, merge_lists=merge_lists) if extras: - ex = safe_merge(Configuration(extras), ex) + ex = safe_merge(Configuration(extras), ex, merge_lists=merge_lists) return Configuration(ex) @@ -389,7 +396,7 @@ return result -def merge(user, default, filter_comments=True, extend_lists=False): +def merge(user, default, filter_comments=True, merge_lists=None): """Logically merge the configuration in `user` into `default`. :param ~configmix.config.Configuration user: @@ -399,9 +406,10 @@ the base configuration where `user` is logically merged into :param bool filter_comments: flag whether to filter comment keys that start with any of the items in :data:`.COMMENTS` - :param bool extend_lists: When ``True`` then lists will be - extended instead of overwritten by the - merge process + :param merge_lists: When ``None`` then lists will be overwritten + by the merge process. When ``extend`` then + lists will be extended instead. + :type merge_lists: str or None :returns: `user` with the necessary amendments from `default`. If `user` is ``None`` then `default` is returned. @@ -438,14 +446,25 @@ # do not copy del user[k] else: - user[k] = _merge(ukv, v, filter_comments, extend_lists) + user[k] = _merge(ukv, v, filter_comments, merge_lists) else: user[k] = v + elif merge_lists is not None: + if merge_lists == "extend": + if isinstance(user, list) and isinstance(default, list): + for idx, value in enumerate(default): + user.insert(idx, value) + elif merge_lists == "prepend": + if isinstance(user, list) and isinstance(default, list): + user.extend(default) + else: + raise ValueError( + "unknown strategy for merge_lists: %s" % (merge_lists,)) _filter_deletions(user) return user -def _merge(user, default, filter_comments, extend_lists): +def _merge(user, default, filter_comments, merge_lists): """Recursion helper for :func:`.merge` """ @@ -462,13 +481,24 @@ # do not copy del user[k] else: - user[k] = _merge(ukv, v, filter_comments, extend_lists) + user[k] = _merge(ukv, v, filter_comments, merge_lists) else: user[k] = v + elif merge_lists is not None: + if merge_lists == "extend": + if isinstance(user, list) and isinstance(default, list): + for idx, value in enumerate(default): + user.insert(idx, value) + elif merge_lists == "prepend": + if isinstance(user, list) and isinstance(default, list): + user.extend(default) + else: + raise ValueError( + "unknown strategy for merge_lists: %s" % (merge_lists,)) return user -def safe_merge(user, default, filter_comments=True, extend_lists=False): +def safe_merge(user, default, filter_comments=True, merge_lists=None): """A more safe version of :func:`.merge` that makes deep copies of the returned container objects. @@ -499,14 +529,25 @@ # do not copy del user[k] else: - user[k] = _safe_merge(ukv, v, filter_comments, extend_lists) + user[k] = _safe_merge(ukv, v, filter_comments, merge_lists) else: user[k] = copy.deepcopy(v) + elif merge_lists is not None: + if merge_lists == "extend": + if isinstance(user, list) and isinstance(default, list): + for idx, value in enumerate(default): + user.insert(idx, copy.deepcopy(value)) + elif merge_lists == "prepend": + if isinstance(user, list) and isinstance(default, list): + user.extend(copy.deepcopy(default)) + else: + raise ValueError( + "unknown strategy for merge_lists: %s" % (merge_lists,)) _filter_deletions(user) return user -def _safe_merge(user, default, filter_comments, extend_lists): +def _safe_merge(user, default, filter_comments, merge_lists): """Recursion helper for :func:`safe_merge` """ @@ -523,9 +564,20 @@ # do not copy del user[k] else: - user[k] = _safe_merge(ukv, v, filter_comments, extend_lists) + user[k] = _safe_merge(ukv, v, filter_comments, merge_lists) else: user[k] = copy.deepcopy(v) + elif merge_lists is not None: + if merge_lists == "extend": + if isinstance(user, list) and isinstance(default, list): + for idx, value in enumerate(default): + user.insert(idx, copy.deepcopy(value)) + elif merge_lists == "prepend": + if isinstance(user, list) and isinstance(default, list): + user.extend(copy.deepcopy(default)) + else: + raise ValueError( + "unknown strategy for merge_lists: %s" % (merge_lists,)) return user
