asyncio-compatible timeout context manager.
The context manager is useful in cases when you want to apply timeout
logic around block of code or in cases when asyncio.wait_for()
is
not suitable. Also it's much faster than asyncio.wait_for()
because timeout
doesn't create a new task.
The timeout(delay, *, loop=None)
call returns a context manager
that cancels a block on timeout expiring:
async with timeout(1.5): await inner()
- If
inner()
is executed faster than in1.5
seconds nothing happens. - Otherwise
inner()
is cancelled internally by sendingasyncio.CancelledError
into butasyncio.TimeoutError
is raised outside of context manager scope.
timeout parameter could be None
for skipping timeout functionality.
Another possibility is to use timeout
as function decorator:
@timeout(1.5) async def possibly_long_running(): pass
Alternatively, timeout_at(when)
can be used for scheduling
at the absolute time:
loop = asyncio.get_event_loop() now = loop.time() async with timeout_at(now + 1.5): await inner()
Please note: it is not POSIX time but a time with undefined starting base, e.g. the time of the system power on.
Context manager has .expired
property for check if timeout happens
exactly in context manager:
async with timeout(1.5) as cm: await inner() print(cm.expired)
The property is True
if inner()
execution is cancelled by
timeout context manager.
If inner()
call explicitly raises TimeoutError
cm.expired
is False
.
The scheduled deadline time is available as .deadline
property:
async with timeout(1.5) as cm: cm.deadline
Not finished yet timeout can be rescheduled by shift_by()
or shift_to()
methods:
async with timeout(1.5) as cm: cm.shift_by(1) # add another second on waiting cm.shift_to(loop.time() + 5) # reschedule to now+5 seconds
Rescheduling is forbidden if the timeout is expired or after exit from async with
code block.
$ pip install async-timeout
The library is Python 3 only!
The module is written by Andrew Svetlov.
It's Apache 2 licensed and freely available.