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

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

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

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

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

@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

@abstractmethod
def wait_for_slot(self) -> None:
    """
    Wait until a slot becomes.
    """
    pass