Asyncio Queue Worker Pool

import asyncio
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    identifier: int
    payload: str


async def worker(
    name: str,
    queue: "asyncio.Queue[Job | None]",
    results: list[tuple[str, int]],
) -> None:
    while True:
        job = await queue.get()
        try:
            if job is None:
                return
            await asyncio.sleep(0.01)
            results.append((name, job.identifier))
        finally:
            queue.task_done()


async def run_jobs(jobs: list[Job], workers: int = 3) -> list[tuple[str, int]]:
    if workers < 1:
        raise ValueError("workers must be positive")
    queue: asyncio.Queue[Job | None] = asyncio.Queue(maxsize=workers * 2)
    results: list[tuple[str, int]] = []
    tasks = [
        asyncio.create_task(worker(f"worker-{index}", queue, results))
        for index in range(workers)
    ]
    for job in jobs:
        await queue.put(job)
    for _ in tasks:
        await queue.put(None)
    await queue.join()
    await asyncio.gather(*tasks)
    return results


def main() -> None:
    jobs = [Job(index, f"payload-{index}") for index in range(8)]
    results = asyncio.run(run_jobs(jobs))
    print(f"processed {len(results)} jobs")
    for name, identifier in sorted(results, key=lambda row: row[1]):
        print(name, "->", identifier)
    assert sorted(identifier for _, identifier in results) == list(range(8))


if __name__ == "__main__":
    main()