Test: Run many tasks with Prefect

run_tasks.py

"""Run tasks with prefect, using DaskTaskRunner."""
from prefect import flow, task, get_run_logger
from prefect_dask import DaskTaskRunner

import time

@task
def my_task(message, count):
    logger = get_run_logger()
    for i in range(count):
        logger.info(f"Repeated message {i} out of {count} times: {message}")
        time.sleep(1)


@flow
def my_flow(messages):
    logger = get_run_logger()
    print(logger.extra)
    for message in messages:
        logger.info(f"Repeating message: {message}")
        my_task.submit(message, 100)
        # shell_run_command.submit(command=command, return_all=True)


# call the flow!
if __name__ == "__main__":
    messages = ["beepboop"] * 10000
    my_flow(messages)