@task
async def hello_number(nmbr: int):
    log = get_run_logger()
    <http://log.info|log.info>("Hello, my number is: %d", nmbr)
    await asyncio.sleep(1)


@flow(name="example_flow_concurrency_limit")
async def main():

    r = list(range(20))

    async with concurrency_limit(hello_number, 1) as tsk:
        await tsk.map(r)

async def wait_for_concurrency_limit(
    client: OrionClient,
    tag: str,
    check_interval: float = 0.5,
    timeout: float = 0.0
):

    start_time = time.time()

    while True:
        if 0 < timeout < (time.time() - start_time):
            LOG.warning("Concurrency wait timeout")
            break

        tasks = await client.read_task_runs(task_run_filter=TaskRunFilter(tags={"all_": [tag]}))
        active_tasks = [x for x in tasks if x.state.type not in TERMINAL_STATES]

        if len(active_tasks) > 0:
            await anyio.sleep(check_interval)

@asynccontextmanager
async def concurrency_limit(task: Task, limit: int = 10) -> Task:

    context = get_run_context()

    tag = f"{context.flow_run.id}-{task.name}"

    client = get_client()

    await client.create_concurrency_limit(tag, limit)
    decorated_task: Task = task.with_options(tags=[tag])

    try:
        yield decorated_task
        await wait_for_concurrency_limit(client, tag)
    finally:
        await client.delete_concurrency_limit_by_tag(tag)