Graceful asyncio Cancellation with Cleanup

import asyncio
import random
from dataclasses import dataclass


@dataclass
class Worker:
    name: str
    processed: int = 0
    cleaned_up: bool = False

    async def run(self, queue: asyncio.Queue[int | None]) -> None:
        try:
            while True:
                item = await queue.get()
                try:
                    if item is None:
                        return
                    await asyncio.sleep(random.uniform(0.005, 0.02))
                    self.processed += 1
                finally:
                    queue.task_done()
        except asyncio.CancelledError:
            # Cancellation must propagate, but cleanup below still runs.
            raise
        finally:
            self.cleaned_up = True


async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue()
    for value in range(1, 31):
        queue.put_nowait(value)

    workers = [Worker(name=f"worker-{index}") for index in range(3)]
    tasks = [asyncio.create_task(worker.run(queue)) for worker in workers]

    await asyncio.sleep(0.02)
    for task in tasks:
        task.cancel()

    outcomes = await asyncio.gather(*tasks, return_exceptions=True)
    cancelled = sum(isinstance(item, asyncio.CancelledError) for item in outcomes)
    print(f"cancelled={cancelled} processed={sum(w.processed for w in workers)}")
    print(f"cleaned_up={all(w.cleaned_up for w in workers)}")
    print(f"pending={queue.qsize()}")


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