A timeout is a configured duration after which an incomplete operation is treated as exceeding its allowed execution time. In asynchronous applications, it prevents a coroutine from waiting indefinitely, preserving resources and giving users timely feedback. section 2.4 section 2.10
Choose timeout behavior according to whether unfinished work should stop:
- Use asyncio.wait_for(awaitable, timeout=seconds) to raise TimeoutError and automatically request cancellation when the limit expires. section 2.4.2
- Wrap a task with asyncio.shield when the caller should stop waiting but the underlying coroutine must continue running. section 2.4.2
- Use asyncio.wait(tasks, timeout=seconds) for groups: it returns done and pending sets, raises no timeout exception, and leaves pending tasks running. section 4.6.4
- Cancel pending tasks explicitly after asyncio.wait when the timeout should terminate unfinished work. section 4.6.4
Protecting a product response
A backend-for-frontend service can cap the product-service wait at one second and return HTTP 504 on timeout. Optional cart, favorites, and inventory data can become null or be omitted while usable product data still reaches the user. section 10.3.3
Listing 2.12 Creating a timeout for a task with wait_for
import asyncio
from util import delay
async def main():
delay_task = asyncio.create_task(delay(2))
try:
result = await asyncio.wait_for(delay_task, timeout=1)
print(result)
except asyncio.exceptions.TimeoutError:
print('Got a timeout!')
print(f'Was the task cancelled? {delay_task.cancelled()}')
asyncio.run(main())
Run the wait_for example to observe a two-second task timing out after one second, raising TimeoutError, and becoming canceled. section 2.4.2
Listing 4.15 Using timeouts with wait
@async_timed()
async def main():
async with aiohttp.ClientSession() as session:
url = 'https:/ / example .com'
fetchers = [asyncio.create_task(fetch_status(session, url),
asyncio.create_task(fetch_status(session, url),
asyncio.create_task(fetch_status(session, url, delay=3))]
done, pending = await asyncio.wait(fetchers, timeout=1)
print(f'Done task count: {len(done)}')
print(f'Pending task count: {len(pending)}')
for done_task in done:
result = await done_task
print(result)
asyncio.run(main())
Compare group-timeout behavior by inspecting completed and pending fetch tasks after asyncio.wait returns at one second. section 4.6.4
