Skip to content

rate_limit

multifutures.wait 

wait(wait_time: float = 0.1, *, jitter: bool = True) -> None

Wait for wait_time + some random jitter

Source code in multifutures/_rate_limit.py 
13
14
15
16
17
18
19
20
def wait(wait_time: float = 0.1, *, jitter: bool = True) -> None:
    """Wait for `wait_time` + some random `jitter`"""
    if jitter:
        # Set jitter to be <= 1% of wait_time
        jitter_time = random.random() / (1 / wait_time * 100)  # noqa: S311 - suspicious-non-cryptographic-random-usage
    else:
        jitter_time = 0
    time.sleep(wait_time + jitter_time)

multifutures.RateLimit

A wrapper around the limits library. It defaults to using the Moving Window Strategy on the Memory Storage.

Warning

The default strategy only works with multi-threading, not multi-processing. If you want to use it with multiprocessing you need to use a different storage backend (e.g. redis).

Usage

rate_limit = RateLimit()

while rate_limit.reached():
    wait()
PARAMETER DESCRIPTION
rate_limit

The rate limit. An instance of limits.parse. Defaults to 5 requests per seconds.

TYPE: limits.RateLimitItem | None DEFAULT: None

strategy

The strategy to use. Defaults to a Moving Window using the Memory Storage.

TYPE: limits.strategies.RateLimiter | None DEFAULT: None

Source code in multifutures/_rate_limit.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
class RateLimit:
    """
    A wrapper around the [limits](https://limits.readthedocs.io/en/stable/index.html) library.
    It defaults to using the [Moving Window Strategy][limits.strategies.MovingWindowRateLimiter]
    on the [Memory Storage][limits.storage.MemoryStorage].

    Warning:
        The default strategy only works with multi-threading, not multi-processing.
        If you want to use it with multiprocessing you need to use a different storage backend (e.g. redis).

    ## Usage

        rate_limit = RateLimit()

        while rate_limit.reached():
            wait()

    Arguments:
        rate_limit: The rate limit. An instance of [limits.parse][limits.parse].
            Defaults to 5 requests per seconds.
        strategy: The strategy to use. Defaults to a Moving Window using the Memory Storage.

    """

    def __init__(
        self,
        rate_limit: limits.RateLimitItem | None = None,
        strategy: limits.strategies.RateLimiter | None = None,
    ) -> None:
        import limits

        self.rate_limit = rate_limit or limits.parse("5/second")
        if strategy is None:
            storage = limits.storage.MemoryStorage()
            strategy = limits.strategies.MovingWindowRateLimiter(storage)
        self.strategy = strategy

    def reached(self, identifier: str = "") -> bool:
        """
        Returns `True` if the rate limit has been reached, `False` otherwise.

        `RateLimit` instances can be reused in different contexts.
        To do so a unique identifier per-context must be provided.

        Arguments:
            identifier: The identifier allows you to reuse the `RateLimit` instance in different contexts.

        """
        return not self.strategy.hit(
            self.rate_limit,
            identifier,
        )

reached 

reached(identifier: str = '') -> bool

Returns True if the rate limit has been reached, False otherwise.

RateLimit instances can be reused in different contexts. To do so a unique identifier per-context must be provided.

PARAMETER DESCRIPTION
identifier

The identifier allows you to reuse the RateLimit instance in different contexts.

TYPE: str DEFAULT: ''

Source code in multifutures/_rate_limit.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def reached(self, identifier: str = "") -> bool:
    """
    Returns `True` if the rate limit has been reached, `False` otherwise.

    `RateLimit` instances can be reused in different contexts.
    To do so a unique identifier per-context must be provided.

    Arguments:
        identifier: The identifier allows you to reuse the `RateLimit` instance in different contexts.

    """
    return not self.strategy.hit(
        self.rate_limit,
        identifier,
    )