AsyncIO Task Scheduler

AsyncScheduler.py

#!/usr/bin/env python3

import asyncio
from typing import Coroutine, List


class AsyncScheduler:
    @classmethod
    def run(cls, tasks: List[Coroutine], limit: int = 3) -> None:
        """
        Given a list of async function calls, execute it with limited concurrency
        Example:
            async def fn(x):
                ...

            AsyncScheduler.run([fn(1), fn(2), fn(3)], limit=2)
        """
        if len(tasks) == 0:
            return
        asyncio.run(cls.__run(tasks, limit))

    @classmethod
    async def __run(cls, tasks, limit) -> None:
        total_tasks = len(tasks)
        running = set()
        while 1:
            while len(tasks) and len(running) < limit:
                running.add(asyncio.create_task(tasks.pop(0)))
            utils.notice(
                f"processing {len(running)} tasks, "
                f"{len(tasks)} pending, "
                f"{total_tasks - len(tasks) - len(running)} done, "
                f"{total_tasks} total"
            )
            done, running = await asyncio.wait(
                running, return_when=asyncio.FIRST_COMPLETED
            )
            if not len(running) and not len(tasks):
                return

Example.py

import asyncio
import time

from AsyncScheduler import AsyncScheduler

async def f(x):
    print(f'f{x}')
    await asyncio.sleep(x)
    print(x, time.time())

tasks = [f(2), f(2), f(1), f(2)]
AsyncScheduler.run(tasks, 3)
tasks