Function convert_legacy_task

dask/_task_spec.py:203–256

(
    key: KeyType | None,
    task: _T,
    all_keys: Container,
)

def convert_legacy_task(
    key: KeyType | None,
    task: _T,
    all_keys: Container,
) -> GraphNode | _T:
    if isinstance(task, GraphNode):
        return task

    if type(task) is tuple and task and callable(task[0]):
        func, args = task[0], task[1:]
        new_args = []
        new: object
        for a in args:
            if isinstance(a, dict):
                new = Dict(a)
            else:
                new = convert_legacy_task(None, a, all_keys)
            new_args.append(new)
        return Task(key, func, *new_args)
    try:
        if isinstance(task, (int, float, str, tuple)):
            if task in all_keys:
                if key is None:
                    return Alias(task)
                else:
                    return Alias(key, target=task)
    except TypeError:
        # Unhashable
        pass

    if isinstance(task, (list, tuple, set, frozenset)):
        if is_namedtuple_instance(task):
            return _wrap_namedtuple_task(
                key,
                task,
                partial(
                    convert_legacy_task,
                    None,
                    all_keys=all_keys,
                ),
            )
        else:
            parsed_args = tuple(convert_legacy_task(None, t, all_keys) for t in task)
            if any(isinstance(a, GraphNode) for a in parsed_args):
                return Task(key, _identity_cast, *parsed_args, typ=type(task))
            else:
                return cast(_T, type(task)(parsed_args))
    elif _is_dask_future(task):
        if key is None:
            return Alias(task.key)  # type: ignore[attr-defined]
        else:
            return Alias(key, target=task.key)  # type: ignore[attr-defined]
    else:
        return task


def convert_legacy_graph(