Динамический асинхронный диспетчер задач

Условие

Необходимо построить асинхронный диспетчер, который позволяет в режиме реального времени добавлять новые корутины с указанием задержки их выполнения. Диспетчер должен правильно управлять запуском задач, фиксировать их окончание и предоставлять механизм ожидания завершения как уже запущенных, так и только что добавленных задач.

import asyncio
import random
from typing import Callable, Coroutine, Any

class AsyncTaskScheduler:
    async def add_task(self, task_func: Callable[[], Coroutine[Any, Any, Any]]) -> asyncio.Task:
        """
        Добавляет новую задачу в планировщик.

        :param task_func: Короутина, которая будет выполнена
        :return: Объект asyncio.Task
        """
        pass

    async def wait_for_all_tasks(self):
        """
        Ожидает завершения всех текущих и добавляемых задач.
        """
        pass

async def example_task(task_id: int, duration: float):
    """
    Пример задачи с переменной длительностью выполнения.

    :param task_id: Уникальный идентификатор задачи
    :param duration: Время выполнения задачи
    """
    print(f"Task {task_id} started")
    await asyncio.sleep(duration)
    print(f"Task {task_id} completed after {duration} seconds")
    return task_id

async def main():
    pass

if __name__ == "__main__":
    asyncio.run(main())

Ответ

Храните динамический набор задач и отдельный сигнал окончания приёма. Ожидание завершается только после закрытия планировщика и выполнения всех принятых задач; иначе будущие добавления предсказать невозможно.

Задержку добавим необязательным аргументом. close() вызывается после завершения всех производителей задач; планировщик работает в одном event loop:

import asyncio
import math

class AsyncTaskScheduler:
    def __init__(self):
        self.pending = set()
        self.tasks = []
        self.closed = False
        self.changed = asyncio.Event()

    async def add_task(self, task_func, delay=0):
        if self.closed:
            raise RuntimeError('Scheduler closed')
        if not math.isfinite(delay) or delay < 0:
            raise ValueError('Invalid delay')
        task = asyncio.create_task(self._run(task_func, delay))
        self.pending.add(task)
        self.tasks.append(task)
        task.add_done_callback(self._done)
        return task

    async def _run(self, factory, delay):
        await asyncio.sleep(delay)
        return await factory()

    def _done(self, task):
        self.pending.discard(task)
        self.changed.set()

    async def close(self):
        self.closed = True
        self.changed.set()

    async def wait_for_all_tasks(self):
        while not self.closed or self.pending:
            self.changed.clear()
            await self.changed.wait()
        return await asyncio.gather(*self.tasks, return_exceptions=True)

Callback учитывает и отмену задачи до начала её корутины. Фабрика вызывается после задержки. gather возвращает результаты или исключения в порядке добавления. Отмена ожидания до gather не отменяет рабочие задачи: для общей остановки нужен отдельный контракт.