Class Dict

dask/_task_spec.py:935–1003

class Dict(NestedContainer, Mapping):
    klass = dict

    def __init__(self, /, *args: Any, **kwargs: Any):
        if args:
            assert not kwargs
            if len(args) == 1:
                args = args[0]
                if isinstance(args, dict):  # type: ignore[unreachable]
                    args = tuple(itertools.chain(*args.items()))  # type: ignore[unreachable]
                elif isinstance(args, (list, tuple)):
                    if all(
                        len(el) == 2 if isinstance(el, (list, tuple)) else False
                        for el in args
                    ):
                        args = tuple(itertools.chain(*args))
                else:
                    raise ValueError("Invalid argument provided")

            if len(args) % 2 != 0:
                raise ValueError("Invalid number of arguments provided")

        elif kwargs:
            assert not args
            args = tuple(itertools.chain(*kwargs.items()))

        super().__init__(*args)

    def __repr__(self):
        values = ", ".join(f"{k}: {v}" for k, v in batched(self.args, 2, strict=True))
        return f"Dict({values})"

    def substitute(
        self, subs: dict[KeyType, KeyType | GraphNode], key: KeyType | None = None
    ) -> Dict:
        subs_filtered = {
            # subs can be as large as the whole graph; do not iterate over it!
            k: v
            for k in (self.dependencies & subs.keys())
            if (v := subs[k]) != k
        }
        if not subs_filtered:
            return self

        new_args = []
        for arg in self.args:
            new_arg = (
                arg.substitute(subs_filtered)
                if isinstance(arg, (GraphNode, TaskRef))
                else arg
            )
            new_args.append(new_arg)
        return type(self)(new_args)

    def __iter__(self):
        yield from self.args[::2]

    def __len__(self):