Python Async Queue Runner | Cododel
CODODELDEV
EN / RU
Back to Deck
[utility]

Python Async Queue Runner

SOURCE Python
VERSION 1.0
AUTHOR Cododel

A utility class for creating a queue of asynchronous tasks with a chainable interface. Allows sequentially adding tasks, setting arguments, and running them one after another.

Key Features

  • Chainable interface for convenient task addition
  • Sequential execution of all queued tasks
  • Support for positional and keyword arguments
  • Simple and clear structure

Usage

import asyncio
import random
from services.QueueRunner import QueueRunner


async def task(val):
    sleep = random.uniform(0.05, 1.0)
    print(f'{val} sleeping for {sleep:.2f} seconds')
    await asyncio.sleep(sleep)
    print(f'{val} woke up')


async def main():
    queue = QueueRunner()


    # Add 10 tasks to the queue with chainable interface
    for i in range(10):
        queue.task(task).args(i)


    # Execute the queue sequentially
    await queue.run()


asyncio.run(main())

Source Code

AsyncQueueRunner.py

from typing import List, Callable




class QueueTask:
    task: Callable
    _args: tuple = tuple()
    _kwargs: dict = dict()


    def __init__(self, task: Callable):
        self.task = task


    def args(self, *args, **kwargs):
        self._args = args
        self._kwargs = kwargs
        return self


    async def run(self):
        await self.task(*self._args, **self._kwargs)




class QueueRunner:
    queue: List[QueueTask] = []


    def task(self, task: Callable):
        self.queue.append(QueueTask(task))
        return self


    def args(self, *args, **kwargs):
        self.queue[-1].args(*args, **kwargs)
        return self


    async def run(self):
        for task in self.queue:
            await task.run()
[ ▲ 0 ]