from typing import List
import dask_kubernetes
from dask_kubernetes import make_pod_spec
from kubernetes.client import V1Pod
import prefect
from prefect import flow, get_run_logger, task
from prefect_dask.task_runners import DaskTaskRunner
import time


@task
def say_hello(name: str) -> None:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"hello {name}")


@task
def say_goodbye(name: str) -> None:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"goodbye {name}")


def dask_pod_spec() -> V1Pod:
    return make_pod_spec(
        image="<http://ghcr.io/dask/dask:latest|ghcr.io/dask/dask:latest>",
        memory_limit="1G",
        memory_request="1G",
        cpu_limit=1,
        cpu_request=1,
        extra_pod_config={"serviceAccountName": "default"},
    )


@flow(
    task_runner=DaskTaskRunner(
        cluster_class=dask_kubernetes.KubeCluster,
        cluster_kwargs={"pod_template": dask_pod_spec()},
        adapt_kwargs={"minimum": 1, "maximum": 2},
    )
)
def greetings(names: List[str]) -> None:
    for name in names:
        say_hello(name)
        say_goodbye(name)
    time.sleep(120)


if __name__ == "__main__":
    print(prefect.__version__)
    greetings(["arthur", "trillian", "ford", "marvin"])