Skip to content
Back to top

# Source code for xonsh.lib.collections

    """Base class for chaining DBs"""

    import itertools
    import typing as tp
    from collections import ChainMap
    from collections.abc import MutableMapping, MutableSequence, MutableSet

    class ChainDBDefaultType:
        """Singleton for representing when no default value is given."""

        __inst: tp.Optional["ChainDBDefaultType"] = None

        def __new__(cls):
            if ChainDBDefaultType.__inst is None:
                ChainDBDefaultType.__inst = object.__new__(cls)
            return ChainDBDefaultType.__inst

    ChainDBDefault = ChainDBDefaultType()

    class ChainDB(ChainMap):
        """A ChainMap who's ``_getitem__`` returns either a ChainDB or
        the result. The results resolve to the outermost mapping."""

        def __getitem__(self, key):
            res = None
            results = []
            # Try to get all the data from all the mappings
            for mapping in self.maps:
                results.append(mapping.get(key, ChainDBDefault))
            # if all the results are mapping create a ChainDB
            if all(isinstance(result, MutableMapping) for result in results):
                for result in results:
                    if res is None:
                        res = ChainDB(result)
                    else:
                        res.maps.append(result)
            elif all(
                isinstance(result, MutableSequence | MutableSet) for result in results
            ):
                results_chain = itertools.chain(*results)
                # if all reults have the same type, cast into that type
                if all(isinstance(result, type(results[0])) for result in results):
                    return type(results[0])(results_chain)
                else:
                    return list(results_chain)
            else:
                for result in reversed(results):
                    if result is not ChainDBDefault:
                        return result
                raise KeyError(f"{key} is none of the current mappings")
            return res

        def __setitem__(self, key, value):
            if key not in self:
                super().__setitem__(key, value)
            else:
                # Try to get all the data from all the mappings
                for mapping in reversed(self.maps):
                    if key in mapping:
                        mapping[key] = value

    def _convert_to_dict(cm):
        if isinstance(cm, ChainMap | ChainDB):
            r = {}
            for k, v in cm.items():
                r[k] = _convert_to_dict(v)
            return r
        else:
            return cm