GeezerVison

The World Wide Web, restored to 800 pixels of glory.

GeezerVison is showing cododel.dev — Python Async Queue Runner | Cododel

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()

You are visitor 01402520 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.