Skip to content

RateLimit

sensei.RateLimit

RateLimit(calls, period)

Bases: IRateLimit

The class that manages rate limiting by maintaining tokens and enforcing rate limits. This class implements a token bucket rate-limiting system.

Example
from sensei import RateLimit, Router

calls, period = 1, 1
rate_limit = RateLimit(calls, period)
router = Router('https://example-api.com', rate_limit=rate_limit)

@router.get('/users/{id_}')
def get_user(id_: int) -> User:
    pass

for i in range(5):
    get_user(i)  # Here code will be paused for 1 second after each iteration
PARAMETER DESCRIPTION
calls

The maximum number of requests allowed per period.

TYPE: int

period

The time period in seconds for the rate limit.

TYPE: int

Source code in sensei/client/rate_limiter.py
38
39
40
41
42
43
def __init__(self, calls: int, period: int) -> None:
    super().__init__(calls, period)
    self._tokens: int = calls
    self._last_checked: float = time()
    self._async_lock: asyncio.Lock = asyncio.Lock()
    self._thread_lock: threading.Lock = threading.Lock()

period property writable

period

calls property writable

calls

async_wait_for_slot async

async_wait_for_slot()

Asynchronously wait until a slot becomes available by periodically attempting to acquire a token.

Source code in sensei/client/rate_limiter.py
69
70
71
72
async def async_wait_for_slot(self) -> None:
    """Asynchronously wait until a slot becomes available by periodically attempting to acquire a token."""
    while not await self._async_acquire():
        await asyncio.sleep(self._period / self._calls)

wait_for_slot

wait_for_slot()

Synchronously wait until a slot becomes available by periodically attempting to acquire a token.

Source code in sensei/client/rate_limiter.py
84
85
86
87
def wait_for_slot(self) -> None:
    """Synchronously wait until a slot becomes available by periodically attempting to acquire a token."""
    while not self._acquire():
        sleep(self._period / self._calls)

sensei.types.IRateLimit

IRateLimit(calls, period)

Bases: ABC

The interface that can be used to implement a custom rate limiting system.

The following methods have to be implemented:

  • async_wait_for_slot
  • wait_for_slot
Example
from sensei.types import IRateLimit

class CustomLimit(IRateLimit):
    async def async_wait_for_slot(self) -> None:
        ...

    def wait_for_slot(self) -> None:
        ...
PARAMETER DESCRIPTION
calls

The maximum number of requests allowed per period.

TYPE: int

period

The time period in seconds for the rate limit.

TYPE: int

Source code in sensei/types.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, calls: int, period: int) -> None:
    """
    The interface that can be used to implement a custom rate limiting system.

    The following methods have to be implemented:

    - async_wait_for_slot
    - wait_for_slot

    Example:
        ```python
        from sensei.types import IRateLimit

        class CustomLimit(IRateLimit):
            async def async_wait_for_slot(self) -> None:
                ...

            def wait_for_slot(self) -> None:
                ...
        ```

    Args:
        calls (int): The maximum number of requests allowed per period.
        period (int): The time period in seconds for the rate limit.
    """
    self._calls: int = calls
    self._period: int = period

period property writable

period

calls property writable

calls

async_wait_for_slot abstractmethod async

async_wait_for_slot()

Wait until a slot becomes available.

Source code in sensei/types.py
59
60
61
62
63
64
@abstractmethod
async def async_wait_for_slot(self) -> None:
    """
    Wait until a slot becomes available.
    """
    pass

wait_for_slot abstractmethod

wait_for_slot()

Wait until a slot becomes.

Source code in sensei/types.py
66
67
68
69
70
71
@abstractmethod
def wait_for_slot(self) -> None:
    """
    Wait until a slot becomes.
    """
    pass