base

Rate Limit Protocols for Synchronous and Asynchronous Context Managers

AsyncRateLimit

Bases: Protocol

Asynchronous Rate Limit Protocol

Source code in limitor/base.py

class AsyncRateLimit(Protocol):
    """Asynchronous Rate Limit Protocol"""

    def __init__(self, config: Any, max_concurrent: int | None = None) -> None:
        ...

    def acquire(self) -> None:
        """Acquire an item from the rate limit. This method should block until a token is available"""

    def __aenter__(self) -> AsyncRateLimit:
        """Enter the context manager, acquiring resources if necessary

        This method should return an instance of AsyncRateLimit

        Returns:
            An instance of the rate limit context manager
        """

    def __aexit__(self, exc_type: type[BaseException], exc_val: BaseException, exc_tb: TracebackType) -> None:
        """Exit the context manager, releasing any resources if necessary

        Args:
            exc_type: The type of the exception raised, if any
            exc_val: The value of the exception raised, if any
            exc_tb: The traceback object, if any
        """

__aenter__()

Enter the context manager, acquiring resources if necessary

This method should return an instance of AsyncRateLimit

Returns:

Type	Description
AsyncRateLimit	An instance of the rate limit context manager

Source code in limitor/base.py

def __aenter__(self) -> AsyncRateLimit:
    """Enter the context manager, acquiring resources if necessary

    This method should return an instance of AsyncRateLimit

    Returns:
        An instance of the rate limit context manager
    """

__aexit__(exc_type, exc_val, exc_tb)

Exit the context manager, releasing any resources if necessary

Parameters:

Name	Type	Description	Default
exc_type	type[BaseException]	The type of the exception raised, if any	required
exc_val	BaseException	The value of the exception raised, if any	required
exc_tb	TracebackType	The traceback object, if any	required

Source code in limitor/base.py

def __aexit__(self, exc_type: type[BaseException], exc_val: BaseException, exc_tb: TracebackType) -> None:
    """Exit the context manager, releasing any resources if necessary

    Args:
        exc_type: The type of the exception raised, if any
        exc_val: The value of the exception raised, if any
        exc_tb: The traceback object, if any
    """

acquire()

Acquire an item from the rate limit. This method should block until a token is available

Source code in limitor/base.py

def acquire(self) -> None:
    """Acquire an item from the rate limit. This method should block until a token is available"""

SyncRateLimit

Bases: Protocol

Synchronous Rate Limit Protocol

Source code in limitor/base.py

class SyncRateLimit(Protocol):
    """Synchronous Rate Limit Protocol"""

    def __init__(self, config: Any) -> None:
        ...

    def acquire(self) -> None:
        """Acquire an item from the rate limit. This method should block until a token is available"""

    def __enter__(self) -> SyncRateLimit:
        """Enter the context manager, acquiring resources if necessary

        This method should return an instance of SyncRateLimit

        Returns:
            An instance of the rate limit context manager
        """
        return self

    def __exit__(self, exc_type: type[BaseException], exc_val: BaseException, exc_tb: TracebackType) -> None:
        """Exit the context manager, releasing any resources if necessary

        Args:
            exc_type: The type of the exception raised, if any
            exc_val: The value of the exception raised, if any
            exc_tb: The traceback object, if any
        """

__enter__()

Enter the context manager, acquiring resources if necessary

This method should return an instance of SyncRateLimit

Returns:

Type	Description
SyncRateLimit	An instance of the rate limit context manager

Source code in limitor/base.py

def __enter__(self) -> SyncRateLimit:
    """Enter the context manager, acquiring resources if necessary

    This method should return an instance of SyncRateLimit

    Returns:
        An instance of the rate limit context manager
    """
    return self

__exit__(exc_type, exc_val, exc_tb)

Exit the context manager, releasing any resources if necessary

Parameters:

Name	Type	Description	Default
exc_type	type[BaseException]	The type of the exception raised, if any	required
exc_val	BaseException	The value of the exception raised, if any	required
exc_tb	TracebackType	The traceback object, if any	required

Source code in limitor/base.py

def __exit__(self, exc_type: type[BaseException], exc_val: BaseException, exc_tb: TracebackType) -> None:
    """Exit the context manager, releasing any resources if necessary

    Args:
        exc_type: The type of the exception raised, if any
        exc_val: The value of the exception raised, if any
        exc_tb: The traceback object, if any
    """

acquire()

Acquire an item from the rate limit. This method should block until a token is available

Source code in limitor/base.py

def acquire(self) -> None:
    """Acquire an item from the rate limit. This method should block until a token is available"""