mode.utils.times
¶
Time, date and timezone related utilities.
-
mode.utils.times.
Seconds
= typing.Union[datetime.timedelta, float, str]¶ Seconds can be expressed as float or
timedelta
,
-
class
mode.utils.times.
Bucket
(rate: Union[datetime.timedelta, float, str], over: Union[datetime.timedelta, float, str] = 1.0, *, fill_rate: Union[datetime.timedelta, float, str] = None, capacity: Union[datetime.timedelta, float, str] = None, raises: Type[BaseException] = None, loop: asyncio.events.AbstractEventLoop = None)¶ 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
True
if that amount can be poured now, orFalse
if the caller has to wait.If this returns
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
-
rate
: float = None¶
-
capacity
: float = None¶
-
abstract
pour
(tokens: int = 1) → bool¶
-
abstract
expected_time
(tokens: int = 1) → float¶
-
abstract property
tokens
¶
-
property
fill_rate
¶
-
-
class
mode.utils.times.
TokenBucket
(rate: Union[datetime.timedelta, float, str], over: Union[datetime.timedelta, float, str] = 1.0, *, fill_rate: Union[datetime.timedelta, float, str] = None, capacity: Union[datetime.timedelta, float, str] = None, raises: Type[BaseException] = None, loop: asyncio.events.AbstractEventLoop = None)¶ Rate limiting using the token bucket algorithm.
-
pour
(tokens: int = 1) → bool¶
-
expected_time
(tokens: int = 1) → float¶
-
property
tokens
¶
-
-
mode.utils.times.
rate
(r: float) → float¶ Convert rate string (“100/m”, “2/h” or “0.5/s”) to seconds.
-
mode.utils.times.
rate_limit
(rate: float, over: Union[datetime.timedelta, float, str] = 1.0, *, bucket_type: Type[mode.utils.times.Bucket] = mode.utils.times.TokenBucket, raises: Type[BaseException] = None, loop: asyncio.events.AbstractEventLoop = None) → mode.utils.times.Bucket¶ Create rate limiting manager.
-
mode.utils.times.
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”.
- Parameters
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.
-
mode.utils.times.
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”.