def METHOD_NAME(
    client: AsyncDaskClient, cache: bool = False
) -> Callable[[Callable[..., T]], Callable[..., Awaitable[T]]]:
    from dask import delayed
    def _daskify(func: Callable[..., T]) -> Callable[..., Awaitable[T]]:
        if cache:
            func.cache = {}  # type: ignore[attr-defined]
        delayed_func = delayed(func)
        @wraps(func)
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            if cache:
                key = _cache_key(args, kwargs)  # type: ignore[arg-type]
                future = func.cache.get(key)  # type: ignore[attr-defined]
                if future is None:
                    future = client.compute(delayed_func(*args, **kwargs))
                    func.cache[key] = future  # type: ignore[attr-defined]
            else:
                future = client.compute(delayed_func(*args, **kwargs))
            result = await future
            return result
        return wrapper
    return _daskify