asyncio.wait_for applies a time limit while awaiting a coroutine or task. It raises TimeoutError when the limit expires and automatically cancels the awaited operation. section 2.4.2
Use wait_for when an asynchronous operation must not run indefinitely:
- Catch TimeoutError when exceeding the limit is an expected outcome. section 2.4.2
- Pass an existing task when you need to inspect its cancelled() state or access it after timeout. section 2.4.2
- Wrap the task with asyncio.shield when the timeout should report slow progress without cancelling the underlying task. section 2.4.2
- During server shutdown, give pending tasks a short completion window, then handle expected timeouts and terminate. section 3.6.2
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())
A runnable example limits a two-second delay to one second, catches TimeoutError, and verifies that the task was cancelled. section 2.4.2
