one-at-a-time.py

import asyncio
import prefect   # v2.8.7
from prefect.client import get_client
from prefect import flow, get_run_logger, task, unmapped

"""
Usage:

1. Authenticate to Prefect Cloud
2. Save this script and execute, e.g. `python one-at-a-time.py`

The output is log lines like this:

```
$ python one-at-a-time.py
14:31:12.136 | INFO    | prefect.engine - Created flow run 'placid-potoo' for flow 'One at a time test'
14:31:13.879 | INFO    | Flow run 'placid-potoo' - Starting flow for list of 10 items
...
14:31:14.835 | INFO    | Flow run 'placid-potoo' - Created task run 'mytask-0' for task 'mytask'
14:31:14.836 | INFO    | Flow run 'placid-potoo' - Submitted task run 'mytask-0' for execution.
14:31:15.246 | INFO    | Task run 'mytask-9' - Starting mytask for item 9
14:31:15.536 | INFO    | Task run 'mytask-9' - Finished in state Completed()
14:31:45.566 | INFO    | Task run 'mytask-3' - Starting mytask for item 3      # <--- 30 second interval
14:31:45.862 | INFO    | Task run 'mytask-3' - Finished in state Completed()
14:32:15.993 | INFO    | Task run 'mytask-8' - Starting mytask for item 8      # <--- 30 second interval
14:32:16.339 | INFO    | Task run 'mytask-8' - Finished in state Completed()
14:32:46.481 | INFO    | Task run 'mytask-7' - Starting mytask for item 7      # <--- 31 second interval
14:32:46.855 | INFO    | Task run 'mytask-7' - Finished in state Completed()
...
14:35:49.529 | INFO    | Flow run 'placid-potoo' - Finished in state Completed('All states completed.')

```
"""


async def set_concurrency_limit(tag: str, limit: int):
    async with get_client() as client:
        limit_id = await client.create_concurrency_limit(tag=tag, concurrency_limit=limit)


@flow(name="One at a time test")
def myflow():
    number_of_items = 10
    items = list(range(0, number_of_items))

    logger = get_run_logger()
    logger.info(f"Starting flow for list of {len(items)} items")

    results = mytask.map(items)

    return results


@task(tags=["one-at-a-time"])   # Without the matching tag value, tasks run with no delays
def mytask(item: int) -> int:
    logger = get_run_logger()
    logger.info(f"Starting mytask for item {item}")
    return item


if __name__ == "__main__":
    asyncio.run(set_concurrency_limit("one-at-a-time", 1))
    myflow()