def subs(task, key, val):
    """Perform a substitution on a task.

    Examples
    --------
    >>> from flash.core.serve.dag.utils_test import inc
    >>> subs((inc, 'x'), 'x', 1)  # doctest: +ELLIPSIS
    (<function inc at ...>, 1)

    """
    type_task = type(task)
    if not (type_task is tuple and task and callable(task[0])):  # istask(task):
        try:
            if type_task is type(key) and task == key:
                return val
        except Exception:
            pass
        if type_task is list:
            return [subs(x, key, val) for x in task]
        return task
    newargs = []
    for arg in task[1:]:
        type_arg = type(arg)
        if type_arg is tuple and arg and callable(arg[0]):  # istask(task):
            arg = subs(arg, key, val)
        elif type_arg is list:
            arg = [subs(x, key, val) for x in arg]
        elif type_arg is type(key):
            try:
                # Can't do a simple equality check, since this may trigger
                # a FutureWarning from NumPy about array equality
                # https://github.com/dask/dask/pull/2457
                if len(arg) == len(key) and all(type(aa) is type(bb) and aa == bb for aa, bb in zip(arg, key)):
                    arg = val

            except (TypeError, AttributeError):
                # Handle keys which are not sized (len() fails), but are hashable
                if arg == key:
                    arg = val
        newargs.append(arg)
    return task[:1] + tuple(newargs)
Perform a substitution on a task.

    Examples
    --------
    >>> from flash.core.serve.dag.utils_test import inc
    >>> subs((inc, 'x'), 'x', 1)  # doctest: +ELLIPSIS
    (<function inc at ...>, 1)

    
src/flash/core/serve/dag/task.py