Skip to content
Draft
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES/141.feature
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
`timeout` can now also be used as function decorator.
7 changes: 7 additions & 0 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,13 @@ that cancels a block on *timeout* expiring::
*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::

Expand Down
7 changes: 7 additions & 0 deletions async_timeout/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,13 @@ async def __aexit__(
self._do_exit(exc_type)
return None

def __call__(self, f):
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A type hint is required

async def inner(*args, **kargs):
with self:
return await f(*args, **kargs)

return inner

@property
def expired(self) -> bool:
"""Is timeout expired during execution?"""
Expand Down
20 changes: 20 additions & 0 deletions tests/test_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,26 @@ async def long_running_task():
assert canceled_raised, "CancelledError was not raised"


@pytest.mark.asyncio
async def test_timeout_as_decorator():
canceled_raised = False

@timeout(0.01)
async def long_running_task(arg, *, karg):
assert arg == 1
assert karg == 2
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
nonlocal canceled_raised
canceled_raised = True
raise

with pytest.raises(asyncio.TimeoutError):
await long_running_task(1, karg=2)
assert canceled_raised, "CancelledError was not raised"


@pytest.mark.asyncio
async def test_timeout_finish_in_time():
async def long_running_task():
Expand Down