from prefect import flow, task
import time


# 'db' concurrency limit = 50


@task(tags=["db"], timeout_seconds=60)
def range_task(n: int) -> list[int]:
    return list(range(n))


@task(tags=["db"], timeout_seconds=60)
def pwr_task(index: int) -> int:
    time.sleep(0.5)
    return index**2


@task(tags=["db"], timeout_seconds=60)
def add_task(index: int, to_add: int) -> int:
    time.sleep(0.5)
    return index + to_add


@task(tags=["db"], timeout_seconds=60)
def final_task(pwr: int, added: int) -> int:
    time.sleep(0.5)
    return pwr + added


@flow
async def test_flow(n: int, to_adds: list[int] = [1, 2, 3]):

    indexes = range_task(n)

    pwr_futures = pwr_task.map(index=indexes)

    for to_add in to_adds:
        added_futures = add_task.map(index=pwr_futures, to_add=to_add)

        final_futures = final_task.map(pwr=pwr_futures, added=added_futures)


if __name__ == "__main__":
    import asyncio

    asyncio.run(test_flow(300))

Running

mapped

RUNNING

ctrl^C

sync flow calling async tasks,

.submit

async def

DockerContainer