Skip to content

base

Rate Limit Protocols for Synchronous and Asynchronous Context Managers

AsyncRateLimit

Bases: Protocol

Asynchronous Rate Limit Protocol

Parameters:

Name Type Description Default
bucket_config BucketConfig

Configuration for the rate limit

 required
max_concurrent int | None

Maximum number of concurrent requests allowed to acquire capacity

 None
Source code in limitor/base.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class AsyncRateLimit(Protocol):
    """Asynchronous Rate Limit Protocol

    Args:
        bucket_config: Configuration for the rate limit
        max_concurrent: Maximum number of concurrent requests allowed to acquire capacity
    """

    def __init__(self, bucket_config: BucketConfig, max_concurrent: int | None = None) -> None: ...

    async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
        """Acquire an item from the rate limit. This method should block until a token is available

        Args:
            amount: The amount of capacity to acquire, defaults to 1
            timeout: Optional timeout in seconds for the acquire operation
        """

    async 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
        """
        ...  # pylint: disable=unnecessary-ellipsis

    async 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__() async

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 
72
73
74
75
76
77
78
79
80
async 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
    """
    ...  # pylint: disable=unnecessary-ellipsis

__aexit__(exc_type, exc_val, exc_tb) async

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 
82
83
84
85
86
87
88
89
async 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(amount=1, timeout=None) async

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

Parameters:

Name Type Description Default
amount float

The amount of capacity to acquire, defaults to 1

 1
timeout float | None

Optional timeout in seconds for the acquire operation

 None
Source code in limitor/base.py 
64
65
66
67
68
69
70
async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
    """Acquire an item from the rate limit. This method should block until a token is available

    Args:
        amount: The amount of capacity to acquire, defaults to 1
        timeout: Optional timeout in seconds for the acquire operation
    """

HasCapacity

Bases: Protocol

Protocol for objects that have a capacity attribute

Source code in limitor/base.py 
11
12
13
14
15
class HasCapacity(Protocol):  # pylint: disable=too-few-public-methods
    """Protocol for objects that have a capacity attribute"""

    capacity: float
    """Maximum number of items the bucket can hold i.e. number of requests that can be processed at once"""

capacity instance-attribute

Maximum number of items the bucket can hold i.e. number of requests that can be processed at once

SyncRateLimit

Bases: Protocol

Synchronous Rate Limit Protocol

Parameters:

Name Type Description Default
bucket_config BucketConfig

Configuration for the rate limit

 required
Source code in limitor/base.py 
18
19
20
21
22
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
class SyncRateLimit(Protocol):
    """Synchronous Rate Limit Protocol

    Args:
        bucket_config: Configuration for the rate limit
    """

    def __init__(self, bucket_config: BucketConfig) -> None: ...

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

        Args:
            amount: The amount of capacity to acquire, defaults to 1
        """

    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
        """
        ...  # pylint: disable=unnecessary-ellipsis

    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 
34
35
36
37
38
39
40
41
42
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
    """
    ...  # pylint: disable=unnecessary-ellipsis

__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 
44
45
46
47
48
49
50
51
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(amount=1)

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

Parameters:

Name Type Description Default
amount float

The amount of capacity to acquire, defaults to 1

 1
Source code in limitor/base.py 
27
28
29
30
31
32
def acquire(self, amount: float = 1) -> None:
    """Acquire an item from the rate limit. This method should block until a token is available

    Args:
        amount: The amount of capacity to acquire, defaults to 1
    """