asyncio.exceptions.TimeoutError During Gather Operations in Python

Verified FixPython 3.10+Python Native AsyncIO 3.11+Silo: asyncio

Quick Fix / Solution Rapide

Use return_exceptions=True in asyncio.gather() or wrap tasks with asyncio.wait_for(task, timeout=N) to handle individual slow requests gracefully.

Root Cause Analysis

This error occurs when Python executes multiple concurrent coroutines via asyncio.gather() or asyncio.wait() wrapped in a timeout constraint (such as asyncio.wait_for() or async with asyncio.timeout()), and one or more slow tasks fail to complete within the allotted time limit.

1. Slow Concurrent Requests Exceeding Global Deadlines

When fetching data from multiple external APIs via asyncio.gather(api1(), api2(), api3()) with a 5-second timeout, a single hanging endpoint will cause the entire gather call to abort with TimeoutError.

2. Default return_exceptions=False Behavior

By default, asyncio.gather(..., return_exceptions=False) propagates any exception immediately, cancelling remaining sibling tasks.

3. Modern asyncio.timeout Context Manager in Python 3.11+

Python 3.11 introduced async with asyncio.timeout(seconds):, which raises standard TimeoutError when code inside the block exceeds the deadline.

4. Unbounded Network Sockets Without Individual Timeouts

Relying solely on high-level asyncio timeouts without configuring lower-level HTTP/database client timeouts (e.g. aiohttp.ClientTimeout(total=3)).

Reproduction Code (MCVE)

Example: Bug Reproduction

import asyncio

async def slow_worker():
    await asyncio.sleep(5)

async def execute_with_timeout():
    async with asyncio.timeout(0.01):
        await asyncio.gather(slow_worker())

asyncio.run(execute_with_timeout())

Solution 1: Use return_exceptions=True in asyncio.gather

Set return_exceptions=True so gather returns exception objects in the results list rather than terminating sibling coroutines.

Example: Recommended Solution

import asyncio

async def fast_task():
    return 'Fast Result'

async def failing_task():
    raise TimeoutError('Task exceeded deadline')

async def main():
    results = await asyncio.gather(fast_task(), failing_task(), return_exceptions=True)
    for res in results:
        if isinstance(res, Exception):
            print(f'Handled task failure: {res}')
        else:
            print(f'Handled task success: {res}')

asyncio.run(main())

Solution 2: Apply Individual Timeouts with asyncio.wait_for

Wrap each coroutine with its own individual wait_for to isolate timeouts per request.

Example: Alternative Solution

import asyncio

async def safe_fetch(name, delay):
    try:
        return await asyncio.wait_for(asyncio.sleep(delay, result=f'{name} done'), timeout=0.1)
    except TimeoutError:
        return f'{name} timed out (fallback)'

async def main():
    res = await asyncio.gather(safe_fetch('Task1', 0.01), safe_fetch('Task2', 1.0))
    print(f'Results: {res}')

asyncio.run(main())

A common mistake is catching asyncio.TimeoutError instead of built-in TimeoutError. In Python 3.11+, asyncio.TimeoutError is an alias for built-in TimeoutError. Always catch standard TimeoutError. Edge cases occur with task cancellation: when asyncio.timeout expires, it cancels the inner task by injecting CancelledError. If the inner task catches CancelledError and refuses to yield, it blocks the event loop. Contrast this error with asyncio.CancelledError, which is the internal mechanism used to signal cancellation.

Previous Next