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

Code before:
        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__
            try:
                from distributed.utils import iscoroutinefunction

                self._is_coro = iscoroutinefunction(self.func)
            except Exception:
                self._is_coro = False
        return self._is_coro

    @staticmethod
    def fuse(*tasks: Task, key: KeyType | None = None) -> Task:
        leafs = set()
        all_keys = set()
        all_deps: set[KeyType] = set()
        for t in tasks:
            if t.key not in all_deps:
                leafs.add(t.key)
            all_deps.update(t.dependencies)
            all_keys.add(t.key)
            leafs -= t.dependencies
        external_deps = all_deps - set(all_keys)
        if len(leafs) > 1:
            raise ValueError("Cannot fuse tasks with multiple outputs")

        outkey = leafs.pop()

        return Task(
            key or outkey,
            _execute_subgraph,
            DataNode(None, {t.key: t for t in tasks}),
            outkey,
            {k: Alias(k) for k in external_deps},
            {},
        )


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:

Review comment (suggestion):
nit: Isn't `all_keys` already a `set`?
```suggestion
        external_deps = all_deps - all_keys
        
```

Code after:
        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__
            try:
                from distributed.utils import iscoroutinefunction

                self._is_coro = iscoroutinefunction(self.func)
            except Exception:
                self._is_coro = False
        return self._is_coro

    @staticmethod
    def fuse(*tasks: Task, key: KeyType | None = None) -> Task:
        leafs = set()
        all_keys = set()
        all_deps: set[KeyType] = set()
        for t in tasks:
            all_deps.update(t.dependencies)
            all_keys.add(t.key)
        external_deps = all_deps - all_keys
        leafs = all_keys - all_deps
        if len(leafs) > 1:
            raise ValueError("Cannot fuse tasks with multiple outputs")

        outkey = leafs.pop()

        return Task(
            key or outkey,
            _execute_subgraph,
            DataNode(None, {t.key: t for t in tasks}),
            outkey,
            {k: Alias(k) for k in external_deps},
            {},
        )


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:
            return val
        else: