BaseRateLimiter

from pulsefire.ratelimiters import BaseRateLimiter

Bases: ABC

Base rate limiter class.

Inherit this class to implement a rate limiter.

Source code in pulsefire/ratelimiters.py

class BaseRateLimiter(abc.ABC):
    """Base rate limiter class.

    Inherit this class to implement a rate limiter.
    """

    @abc.abstractmethod
    async def acquire(self, invocation: Invocation) -> float:
        """Acquire a wait_for value in seconds.

        | wait_for | action required  |
        | :------: | ---------------- |
        | -1       | Proceed then synchronize. |
        | 0        | Proceed then skip synchronize. |
        | >0       | Wait for value in seconds then acquire again. |
        """

    @abc.abstractmethod
    async def synchronize(self, invocation: Invocation, headers: dict[str, str]) -> None:
        """Synchronize rate limiting headers to index."""

Functions

acquire abstractmethod async

acquire(invocation: Invocation) -> float

Acquire a wait_for value in seconds.

wait_for	action required
-1	Proceed then synchronize.
0	Proceed then skip synchronize.
>0	Wait for value in seconds then acquire again.

Source code in pulsefire/ratelimiters.py

@abc.abstractmethod
async def acquire(self, invocation: Invocation) -> float:
    """Acquire a wait_for value in seconds.

    | wait_for | action required  |
    | :------: | ---------------- |
    | -1       | Proceed then synchronize. |
    | 0        | Proceed then skip synchronize. |
    | >0       | Wait for value in seconds then acquire again. |
    """

synchronize abstractmethod async

synchronize(invocation: Invocation, headers: dict[str, str]) -> None

Synchronize rate limiting headers to index.

Source code in pulsefire/ratelimiters.py

@abc.abstractmethod
async def synchronize(self, invocation: Invocation, headers: dict[str, str]) -> None:
    """Synchronize rate limiting headers to index."""