Repository: dask/dask
File: dask/_task_spec.py (Python)

Code before:
            head = funcname(self.func)
            tail = ")"
            label_size = 40
            args = self.args
            kwargs = self.kwargs
            if args or kwargs:
                label_size2 = int(
                    (label_size - len(head) - len(tail) - len(str(self.key)))
                    // (len(args) + len(kwargs))
                )
            if args:
                if label_size2 > 5:
                    args_repr = ", ".join(repr(t) for t in args)
                else:
                    args_repr = "..."
            else:
                args_repr = ""
            if kwargs:
                if label_size2 > 5:
                    kwargs_repr = ", " + ", ".join(
                        f"{k}={repr(v)}" for k, v in sorted(kwargs.items())
                    )
                else:
                    kwargs_repr = ", ..."
            else:
                kwargs_repr = ""
            self._repr = f"<Task {self.key!r} {head}({args_repr}{kwargs_repr}{tail}>"
        return self._repr

    def __call__(self, values=()):
        self._verify_values(values)
        new_argspec = tuple(
            a(values) if isinstance(a, GraphNode) else a for a in self.args
        )
        if self.kwargs:
            kwargs = {
                k: kw(values) if isinstance(kw, GraphNode) else kw
                for k, kw in self.kwargs.items()
            }
            return self.func(*new_argspec, **kwargs)
        return self.func(*new_argspec)

    def inline(self, dsk) -> Task:
        new_args = _inline_recursively(self.args, dsk)
        new_kwargs = _inline_recursively(self.kwargs, dsk)
        return Task(self.key, self.func, *new_args, **new_kwargs)

    @property
    def is_coro(self):
        if self._is_coro is None:
            # Note: Can't use cached_property on objects without __dict__

Review comment (suggestion):
I ended up adding those convenience classes that implement the required logic to generate the appropriate container. These things are required to pass tasks or references in containers or nested containers. This should be an intentional opt-in for developers. user provided arguments could always be wrapped by this stuff, for example.

Code after:
            None,
            self.to_container,
            *args,
            _dependencies=_dependencies,
            **kwargs,
        )

    @classmethod
    def to_container(cls, *args, **kwargs):
        return cls.klass(args)


class List(NestedContainer):
    klass = list


class Tuple(NestedContainer):
    klass = tuple


class Set(NestedContainer):
    klass = set


class Dict(NestedContainer):
    klass = dict

    def __init__(
        self, /, *args: Any, _dependencies: set | frozenset | None = None, **kwargs: Any
    ):
        if args:
            if len(args) > 1:
                raise ValueError("Dict can only take one positional argument")
            kwargs = args[0]
        elif not kwargs:
            raise ValueError("Dict needs at least one argument")
        super().__init__(
            *(Tuple(*it) for it in kwargs.items()), _dependencies=_dependencies
        )


class DependenciesMapping(MutableMapping):
    def __init__(self, dsk):
        self.dsk = dsk
        self._removed = set()
        # Set a copy of dsk to avoid dct resizing
        self._cache = dsk.copy()
        self._cache.clear()

    def __getitem__(self, key):
        if (val := self._cache.get(key)) is not None: