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

Code before:
    dsk: Mapping[Key, Any],
    dependencies: Mapping[Key, set[Key]] | None = None,
    *,
    return_stats: bool = False,
) -> dict[Key, Order] | dict[Key, int]:
    """Order nodes in dask graph

    This produces an ordering over our tasks that we use to break ties when
    executing.  We do this ahead of time to reduce a bit of stress on the
    scheduler and also to assist in static analysis.

    This currently traverses the graph as a single-threaded scheduler would
    traverse it.

    Examples
    --------
    >>> inc = lambda x: x + 1
    >>> add = lambda x, y: x + y
    >>> dsk = {'a': 1, 'b': 2, 'c': (inc, 'a'), 'd': (add, 'b', 'c')}
    >>> order(dsk)
    {'a': 0, 'c': 1, 'b': 2, 'd': 3}
    """
    if not dsk:
        return {}  # type: ignore

    dsk = dict(dsk).copy()
    expected_len = len(dsk)

    if dependencies is None:
        dependencies_are_copy = True
        dependencies = {k: get_dependencies(dsk, k) for k in dsk}
    else:
        # Below we're removing items from the sets in this dict
        # We need a deepcopy for that but we only want to do this if necessary
        # since this only happens for special cases.
        dependencies_are_copy = False
        dependencies = dict(dependencies)

    dependents = reverse_dict(dependencies)

    leaf_nodes = {k for k, v in dependents.items() if not v}
    root_nodes = {k for k, v in dependencies.items() if not v}

    result: dict[Key, Order | int] = {}

    # Normalize the graph by removing leaf nodes that are not actual tasks, see
    # for instance da.store where the task is merely an alias
    # to multiple keys, i.e. [key1, key2, ...,]

    # Similarly, we are removing root nodes that are pure data tasks. Those task
    # are embedded in the run_spec of a task and are not runnable. We have to

Review comment (suggestion):
The `dict()` constructor already performs a copy and returns a _new_ dictionary.
```suggestion
    dsk = dict(dsk)
```

Code after:
    dsk: Mapping[Key, Any],
    dependencies: Mapping[Key, set[Key]] | None = None,
    *,
    return_stats: bool = False,
) -> dict[Key, Order] | dict[Key, int]:
    """Order nodes in dask graph

    This produces an ordering over our tasks that we use to break ties when
    executing.  We do this ahead of time to reduce a bit of stress on the
    scheduler and also to assist in static analysis.

    This currently traverses the graph as a single-threaded scheduler would
    traverse it.

    Examples
    --------
    >>> inc = lambda x: x + 1
    >>> add = lambda x, y: x + y
    >>> dsk = {'a': 1, 'b': 2, 'c': (inc, 'a'), 'd': (add, 'b', 'c')}
    >>> order(dsk)
    {'a': 0, 'c': 1, 'b': 2, 'd': 3}
    """
    if not dsk:
        return {}  # type: ignore

    dsk = dict(dsk)
    expected_len = len(dsk)

    if dependencies is None:
        dependencies_are_copy = True
        dependencies = {k: get_dependencies(dsk, k) for k in dsk}
    else:
        # Below we're removing items from the sets in this dict
        # We need a deepcopy for that but we only want to do this if necessary
        # since this only happens for special cases.
        dependencies_are_copy = False
        dependencies = dict(dependencies)

    dependents = reverse_dict(dependencies)

    leaf_nodes = {k for k, v in dependents.items() if not v}
    root_nodes = {k for k, v in dependencies.items() if not v}

    result: dict[Key, Order | int] = {}

    # Normalize the graph by removing leaf nodes that are not actual tasks, see
    # for instance da.store where the task is merely an alias
    # to multiple keys, i.e. [key1, key2, ...,]

    # Similarly, we are removing root nodes that are pure data tasks. Those task
    # are embedded in the run_spec of a task and are not runnable. We have to