mode.utils.times

Time, date and timezone related utilities.

Bucket

Bases: AsyncContextManager

Rate limiting state.

A bucket "pours" tokens at a rate of rate per second (or over').

Calling bucket.pour(), pours one token by default, and returns :const:True if that amount can be poured now, or :const:False if the caller has to wait.

If this returns :const:False, it's prudent to either sleep or raise an exception:

if not bucket.pour():
    await asyncio.sleep(bucket.expected_time())

If you want to consume multiple tokens in one go then specify the number:

if not bucket.pour(10):
    await asyncio.sleep(bucket.expected_time(10))

This class can also be used as an async. context manager, but in that case can only consume one tokens at a time:

async with bucket:
    # do something

By default the async. context manager will suspend the current coroutine and sleep until as soon as the time that a token can be consumed.

If you wish you can also raise an exception, instead of sleeping, by providing the raises keyword argument:

# hundred tokens in one second, and async with: raises TimeoutError

class MyError(Exception):
    pass

bucket = Bucket(100, over=1.0, raises=MyError)

async with bucket:
    # do something

Source code in mode/utils/times.py

class Bucket(AsyncContextManager):
    """Rate limiting state.

    A bucket "pours" tokens at a rate of ``rate`` per second (or over').

    Calling `bucket.pour()`, pours one token by default, and returns
    :const:`True` if that amount can be poured now, or :const:`False` if the
    caller has to wait.

    If this returns :const:`False`, it's prudent to either sleep or raise
    an exception:

    ```python
    if not bucket.pour():
        await asyncio.sleep(bucket.expected_time())
    ```

    If you want to consume multiple tokens in one go then specify the number:

    ```python
    if not bucket.pour(10):
        await asyncio.sleep(bucket.expected_time(10))
    ```

    This class can also be used as an async. context manager, but in that case
    can only consume one tokens at a time:

    ```python
    async with bucket:
        # do something
    ```

    By default the async. context manager will suspend the current coroutine
    and sleep until as soon as the time that a token can be consumed.

    If you wish you can also raise an exception, instead of sleeping, by
    providing the `raises` keyword argument:

    ```python
    # hundred tokens in one second, and async with: raises TimeoutError

    class MyError(Exception):
        pass

    bucket = Bucket(100, over=1.0, raises=MyError)

    async with bucket:
        # do something
    ```
    """

    rate: float
    capacity: float

    _tokens: float

    def __init__(
        self,
        rate: Seconds,
        over: Seconds = 1.0,
        *,
        fill_rate: Seconds = None,
        capacity: Seconds = None,
        raises: Optional[Type[BaseException]] = None,
        loop: Optional[asyncio.AbstractEventLoop] = None,
    ) -> None:
        self.rate = want_seconds(rate)
        self.capacity = want_seconds(over)
        self.raises = raises
        self.loop = loop
        self._tokens = self.capacity
        self.__post_init__()

    def __post_init__(self) -> None: ...

    @abc.abstractmethod
    def pour(self, tokens: int = 1) -> bool: ...

    @abc.abstractmethod
    def expected_time(self, tokens: int = 1) -> float: ...

    @property
    @abc.abstractmethod
    def tokens(self) -> float: ...

    @property
    def fill_rate(self) -> float:
        #: Defaults to rate! If you want the bucket to fill up
        #: faster/slower, then just override this.
        return self.rate

    async def __aenter__(self) -> "Bucket":
        if not self.pour():
            if self.raises:
                raise self.raises()
            expected_time = self.expected_time()
            await asyncio.sleep(expected_time)
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]] = None,
        exc_val: Optional[BaseException] = None,
        exc_tb: Optional[TracebackType] = None,
    ) -> Optional[bool]:
        return None

TokenBucket

Bases: Bucket

Rate limiting using the token bucket algorithm.

Source code in mode/utils/times.py

class TokenBucket(Bucket):
    """Rate limiting using the token bucket algorithm."""

    _tokens: float
    _last_pour: float

    def __post_init__(self) -> None:
        self._last_pour = TIME_MONOTONIC()

    def pour(self, tokens: int = 1) -> bool:
        need = tokens
        have = self.tokens
        if have < need:
            return False
        self._tokens -= tokens
        return True

    def expected_time(self, tokens: int = 1) -> float:
        have = self._tokens
        need = max(tokens, have)
        time_left = (need - have) / self.fill_rate
        return max(time_left, 0.0)

    @property
    def tokens(self) -> float:
        now = TIME_MONOTONIC()
        if now < self._last_pour:
            return self._tokens
        if self._tokens < self.capacity:
            delta = self.fill_rate * (now - self._last_pour)
            self._tokens = min(self.capacity, self._tokens + delta)
            self._last_pour = now
        return self._tokens

humanize_seconds(secs, *, prefix='', suffix='', sep='', now='now', microseconds=False)

Show seconds in human form.

For example, 60 becomes "1 minute", and 7200 becomes "2 hours".

Parameters:

Name	Type	Description	Default
secs	float	Seconds to format (as float or int).	required
prefix	str	can be used to add a preposition to the output (e.g., 'in' will give 'in 1 second', but add nothing to 'now').	''
suffix	str	same as prefix, adds suffix unless 'now'.	''
sep	str	separator between prefix and number.	''
now	str	Literal 'now'.	'now'
microseconds	bool	Include microseconds.	False

Source code in mode/utils/times.py

def humanize_seconds(
    secs: float,
    *,
    prefix: str = "",
    suffix: str = "",
    sep: str = "",
    now: str = "now",
    microseconds: bool = False,
) -> str:
    """Show seconds in human form.

    For example, 60 becomes "1 minute", and 7200 becomes "2 hours".

    Arguments:
        secs: Seconds to format (as `float` or `int`).
        prefix (str): can be used to add a preposition to the output
            (e.g., 'in' will give 'in 1 second', but add nothing to 'now').
        suffix (str): same as prefix, adds suffix unless 'now'.
        sep (str): separator between prefix and number.
        now (str): Literal 'now'.
        microseconds (bool): Include microseconds.
    """
    secs = float(format(float(secs), ".2f"))
    for unit, divider, formatter in TIME_UNITS:
        if secs >= divider:
            w = secs / float(divider)
            return f"{prefix}{sep}{formatter(w)} {pluralize(int(w), unit)}{suffix}"
    if microseconds and secs > 0.0:
        return f"{prefix}{sep}{secs:.2f} seconds{suffix}"
    return now

humanize_seconds_ago(secs, *, prefix='', suffix=' ago', sep='', now='just now', microseconds=False)

Show seconds in "3.33 seconds ago" form.

If seconds are less than one, returns "just now".

Source code in mode/utils/times.py

def humanize_seconds_ago(
    secs: float,
    *,
    prefix: str = "",
    suffix: str = " ago",
    sep: str = "",
    now: str = "just now",
    microseconds: bool = False,
) -> str:
    """Show seconds in "3.33 seconds ago" form.

    If seconds are less than one, returns "just now".
    """
    return humanize_seconds(
        secs,
        prefix=prefix,
        suffix=suffix,
        sep=sep,
        now=now,
        microseconds=microseconds,
    )

rate(r)

Convert rate string ("100/m", "2/h" or "0.5/s") to seconds.

Source code in mode/utils/times.py

@singledispatch
def rate(r: float) -> float:
    """Convert rate string (`"100/m"`, `"2/h"` or `"0.5/s"`) to seconds."""
    return r

rate_limit(rate, over=1.0, *, bucket_type=TokenBucket, raises=None, loop=None)

Create rate limiting manager.

Source code in mode/utils/times.py

def rate_limit(
    rate: float,
    over: Seconds = 1.0,
    *,
    bucket_type: Type[Bucket] = TokenBucket,
    raises: Optional[Type[BaseException]] = None,
    loop: Optional[asyncio.AbstractEventLoop] = None,
) -> Bucket:
    """Create rate limiting manager."""
    return bucket_type(rate, over, raises=raises, loop=loop)

want_seconds(s)

Convert Seconds to float.

Source code in mode/utils/times.py

@singledispatch
def want_seconds(s: float) -> float:
    """Convert `Seconds` to float."""
    return s