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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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
        """

    def acquire_ctx(
        self, amount: float = 1, timeout: float | None = None
    ) -> AbstractAsyncContextManager[AsyncRateLimit]:
        """Parameterized async context manager to acquire a specified amount of capacity

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

        Returns:
            An async context manager yielding the rate limit instance
        """
        ...  # pyrefly: ignore

    async def reconcile(self, actual: float, estimated: float) -> None:
        """Reconcile the actual tokens/capacity used against the estimated amount

        Args:
            actual: The actual amount of tokens/capacity used
            estimated: The estimated amount of tokens/capacity previously acquired
        """

    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
        """
        ...  # pyrefly: ignore

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> 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 
121
122
123
124
125
126
127
128
129
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
    """
    ...  # pyrefly: ignore

__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] | None

The type of the exception raised, if any

 required
exc_val BaseException | None

The value of the exception raised, if any

 required
exc_tb TracebackType | None

The traceback object, if any

 required
Source code in limitor/base.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> 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 
91
92
93
94
95
96
97
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
    """

acquire_ctx(amount=1, timeout=None)

Parameterized async context manager to acquire a specified amount of capacity

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

Returns:

Type Description
AbstractAsyncContextManager[AsyncRateLimit]

An async context manager yielding the rate limit instance
Source code in limitor/base.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
def acquire_ctx(
    self, amount: float = 1, timeout: float | None = None
) -> AbstractAsyncContextManager[AsyncRateLimit]:
    """Parameterized async context manager to acquire a specified amount of capacity

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

    Returns:
        An async context manager yielding the rate limit instance
    """
    ...  # pyrefly: ignore

reconcile(actual, estimated) async

Reconcile the actual tokens/capacity used against the estimated amount

Parameters:

Name Type Description Default
actual float

The actual amount of tokens/capacity used

 required
estimated float

The estimated amount of tokens/capacity previously acquired

 required
Source code in limitor/base.py 
113
114
115
116
117
118
119
async def reconcile(self, actual: float, estimated: float) -> None:
    """Reconcile the actual tokens/capacity used against the estimated amount

    Args:
        actual: The actual amount of tokens/capacity used
        estimated: The estimated amount of tokens/capacity previously acquired
    """

HasCapacity

Bases: Protocol

Protocol for objects that have a capacity attribute

Source code in limitor/base.py 
12
13
14
15
16
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 
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
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 acquire_ctx(self, amount: float = 1) -> AbstractContextManager[SyncRateLimit]:
        """Parameterized context manager to acquire a specified amount of capacity

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

        Returns:
            A context manager yielding the rate limit instance
        """
        ...  # pyrefly: ignore

    def reconcile(self, actual: float, estimated: float) -> None:
        """Reconcile the actual tokens/capacity used against the estimated amount

        Args:
            actual: The actual amount of tokens/capacity used
            estimated: The estimated amount of tokens/capacity previously acquired
        """

    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
        """
        ...  # pyrefly: ignore

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> 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 
54
55
56
57
58
59
60
61
62
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
    """
    ...  # pyrefly: ignore

__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] | None

The type of the exception raised, if any

 required
exc_val BaseException | None

The value of the exception raised, if any

 required
exc_tb TracebackType | None

The traceback object, if any

 required
Source code in limitor/base.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> 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 
28
29
30
31
32
33
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
    """

acquire_ctx(amount=1)

Parameterized context manager to acquire a specified amount of capacity

Parameters:

Name Type Description Default
amount float

The amount of capacity to acquire, defaults to 1

 1

Returns:

Type Description
AbstractContextManager[SyncRateLimit]

A context manager yielding the rate limit instance
Source code in limitor/base.py 
35
36
37
38
39
40
41
42
43
44
def acquire_ctx(self, amount: float = 1) -> AbstractContextManager[SyncRateLimit]:
    """Parameterized context manager to acquire a specified amount of capacity

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

    Returns:
        A context manager yielding the rate limit instance
    """
    ...  # pyrefly: ignore

reconcile(actual, estimated)

Reconcile the actual tokens/capacity used against the estimated amount

Parameters:

Name Type Description Default
actual float

The actual amount of tokens/capacity used

 required
estimated float

The estimated amount of tokens/capacity previously acquired

 required
Source code in limitor/base.py 
46
47
48
49
50
51
52
def reconcile(self, actual: float, estimated: float) -> None:
    """Reconcile the actual tokens/capacity used against the estimated amount

    Args:
        actual: The actual amount of tokens/capacity used
        estimated: The estimated amount of tokens/capacity previously acquired
    """