Implement timeout.at(when)

README.rst

@@ -22,7 +22,7 @@ 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(timeout, *, loop=None)`` call returns a context manager
+The ``timeout(delay, *, loop=None)`` call returns a context manager
 that cancels a block on *timeout* expiring::
 
    async with timeout(1.5):
@@ -37,6 +37,20 @@ that cancels a block on *timeout* expiring::
 *timeout* parameter could be ``None`` for skipping timeout functionality.
 
 
+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_timeout/__init__.py

@@ -3,11 +3,13 @@
 
 from types import TracebackType
 from typing import Optional, Type, Any  # noqa
+from typing_extensions import final
 
 
 __version__ = '3.0.1'
 
 
+@final
 class timeout:
     """timeout context manager.
 
@@ -22,9 +24,24 @@ class timeout:
     timeout - value in seconds or None to disable timeout logic
     loop - asyncio compatible event loop
     """
+    @classmethod
+    def at(cls, when: float) -> 'timeout':
+        """Schedule the timeout at absolute time.
+
+        when arguments points on the time in the same clock system
+        as loop.time().
+
+        Please note: it is not POSIX time but a time with
+        undefined starting base, e.g. the time of the system power on.
+
+        """
+        ret = cls(None)
+        ret._cancel_at = when
+        return ret
+
     def __init__(self, timeout: Optional[float],
                  *, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
-        self._timeout = timeout
+        self._delay = timeout
         if loop is None:
             loop = asyncio.get_event_loop()
         self._loop = loop
@@ -56,10 +73,12 @@ async def __aexit__(self,
 
     @property
     def expired(self) -> bool:
+        """Is timeout expired during execution?"""
         return self._cancelled
 
     @property
     def remaining(self) -> Optional[float]:
+        """Number of seconds remaining to the timeout expiring."""
         if self._cancel_at is None:
             return None
         elif self._exited_at is None:
@@ -69,6 +88,12 @@ def remaining(self) -> Optional[float]:
 
     @property
     def elapsed(self) -> float:
+        """Number of elapsed seconds.
+
+        The time is counted starting from entering into
+        the timeout context manager.
+
+        """
         if self._started_at is None:
             return 0.0
         elif self._exited_at is None:
@@ -79,20 +104,27 @@ def elapsed(self) -> float:
     def _do_enter(self) -> 'timeout':
         # Support Tornado 5- without timeout
         # Details: https://github.com/python/asyncio/issues/392
-        if self._timeout is None:
+        if self._delay is None and self._cancel_at is None:
             return self
 
         self._task = _current_task(self._loop)
         if self._task is None:
             raise RuntimeError('Timeout context manager should be used '
                                'inside a task')
 
-        if self._timeout <= 0:
-            self._loop.call_soon(self._cancel_task)
-            return self
-
         self._started_at = self._loop.time()
-        self._cancel_at = self._started_at + self._timeout
+
+        if self._delay is not None:
+            # relative timeout mode
+            if self._delay <= 0:
+                self._loop.call_soon(self._cancel_task)
+                return self
+
+            self._cancel_at = self._started_at + self._delay
+        else:
+            # absolute timeout
+            assert self._cancel_at is not None
+
         self._cancel_handler = self._loop.call_at(
             self._cancel_at, self._cancel_task)
         return self
@@ -103,7 +135,7 @@ def _do_exit(self, exc_type: Type[BaseException]) -> None:
             self._cancel_handler = None
             self._task = None
             raise asyncio.TimeoutError
-        if self._timeout is not None and self._cancel_handler is not None:
+        if self._cancel_handler is not None:
             self._cancel_handler.cancel()
             self._cancel_handler = None
         self._task = None

tests/test_timeout.py

@@ -276,3 +276,22 @@ async def test_timeout_elapsed():
             assert cm.elapsed >= 0.1
             await asyncio.sleep(0.5)
     assert cm.elapsed >= 0.1
+
+
+@pytest.mark.asyncio
+async def test_timeout_at():
+    loop = asyncio.get_event_loop()
+    with pytest.raises(asyncio.TimeoutError):
+        now = loop.time()
+        async with timeout.at(now + 0.01) as cm:
+            await asyncio.sleep(10)
+    assert cm.expired
+
+
+@pytest.mark.asyncio
+async def test_timeout_at_not_fired():
+    loop = asyncio.get_event_loop()
+    now = loop.time()
+    async with timeout.at(now + 1) as cm:
+        await asyncio.sleep(0)
+    assert not cm.expired