Skip to content

Migrate to asyncio.TaskGroup for batch processing once Python 3.10 is dropped #598

@vdusek

Description

@vdusek

Summary

Once we drop support for Python 3.10, we should migrate the batch processing in request_queue.py to use asyncio.TaskGroup (available in Python 3.11+) for better error handling and cleaner code.

Current Implementation

The current batch processing pattern (around lines 855-870) uses manual task management:

tasks = set[asyncio.Task]()
for i in range(max_parallel):
    task = asyncio.create_task(coro, name=f'worker_{i}')
    tasks.add(task)
await asyncio_queue.join()
for task in tasks:
    task.cancel()
results = await asyncio.gather(*tasks)

Proposed Improvement

Use asyncio.TaskGroup for cleaner code and automatic exception handling:

async with asyncio.TaskGroup() as tg:
    workers = [
        tg.create_task(self._batch_add_requests_worker(queue, params))
        for _ in range(max_parallel)
    ]
    await asyncio_queue.join()
    for worker in workers:
        worker.cancel()
# TaskGroup automatically waits and handles exceptions properly

Benefits

  • Better error handling: TaskGroup automatically propagates exceptions from any task
  • Cleaner code: Context manager pattern ensures proper cleanup
  • More Pythonic: Uses modern Python async patterns
  • Automatic waiting: No need for explicit asyncio.gather() call

Prerequisites

  • Drop support for Python 3.10 (TaskGroup requires Python 3.11+)

Metadata

Metadata

Assignees

No one assigned

    Labels

    t-toolingIssues with this label are in the ownership of the tooling team.

    Type

    No type

    Projects

    No projects

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions