Skip to content

core

A simple synchronous implementation of the Generic Cell Rate Algorithm (GCRA)

References: - https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm - https://en.wikipedia.org/wiki/Leaky_bucket

AsyncLeakyBucketGCRA

Continuous-state Leaky Bucket Rate Limiter

Parameters:

Name Type Description Default
bucket_config BucketConfig | None

Configuration for the GCR algorithm with the max capacity and time period in seconds

 None
max_concurrent int | None

Maximum number of concurrent requests allowed to acquire capacity

 None
Note

This implementation is asynchronous and supports bursts up to the capacity within the specified time period

References

https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm

Source code in limitor/generic_cell_rate/core.py 
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
class AsyncLeakyBucketGCRA:
    """Continuous-state Leaky Bucket Rate Limiter

    Args:
        bucket_config: Configuration for the GCR algorithm with the max capacity and time period in seconds
        max_concurrent: Maximum number of concurrent requests allowed to acquire capacity

    Note:
        This implementation is asynchronous and supports bursts up to the capacity within the specified time period

    References:
        https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm
    """

    def __init__(
        self,
        bucket_config: BucketConfig | None = None,
        max_concurrent: int | None = None,
    ):
        # import config and set attributes
        config = bucket_config or BucketConfig()
        self.capacity = config.capacity
        self.seconds = config.seconds

        self.leak_rate = self.capacity / self.seconds  # units per second
        self.T = 1 / self.leak_rate  # time to leak one unit

        # burst rate, but can't do this if the amount is variable
        # self.tau = self.T * self.burst

        self._bucket_level = 0.0  # current volume in the bucket
        self._last_leak: float | None = None  # same as last conforming time or LCT

        self.max_concurrent = max_concurrent
        self._lock = asyncio.Lock()
        self._semaphore: asyncio.Semaphore | AbstractAsyncContextManager[Any] | None = (
            None
        )

    async def _acquire_logic(self, amount: float = 1) -> None:
        """Acquire resources, blocking if necessary to conform to the rate limit

        Args:
            amount: The amount of capacity to check for, defaults to 1

        Notes:
            Adding a lock here ensures that the acquire logic is atomic, but it also means that the
                requests are going to be done in the order they were received  i.e. not out-of-order like
                most async programs.
            The benefit is that with multiple concurrent requests, we can ensure that the bucket level
                is updated correctly and that we don't have multiple requests trying to update the bucket level
                at the same time, which could lead to an inconsistent state i.e. a race condition.
        """
        async with (
            self._lock
        ):  # ensure atomicity given we can have multiple concurrent requests
            t_a = time.monotonic()
            if self._last_leak is None:
                # first cell
                self._bucket_level = 0
                self._last_leak = t_a

            elapsed = t_a - self._last_leak
            self._bucket_level = self._bucket_level - elapsed

            # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
            tau = self.T * (self.capacity - amount)
            if self._bucket_level > tau:
                delay = self._bucket_level - tau
                await asyncio.sleep(delay)

                self._bucket_level = self._bucket_level - delay
                t_a += delay

            self._bucket_level = max(0.0, self._bucket_level) + amount * self.T
            self._last_leak = t_a

    async def _semaphore_acquire(self, amount: float = 1) -> None:
        """Acquire capacity using a semaphore to limit concurrency.

        Args:
            amount: The amount of capacity to acquire, defaults to 1
        """
        if self._semaphore is None:
            self._semaphore = (
                asyncio.Semaphore(self.max_concurrent)
                if self.max_concurrent
                else nullcontext()
            )

        async with self._semaphore:
            await self._acquire_logic(amount)

    async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
        """Acquire capacity, waiting asynchronously until allowed.

        Supports timeout and cancellation.

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

        Raises:
            TimeoutError: If the acquire operation times out after the specified timeout period
        """
        validate_amount(self, amount=amount)

        if timeout is not None:
            try:
                await asyncio.wait_for(self._semaphore_acquire(amount), timeout=timeout)
            except TimeoutError as error:
                raise TimeoutError(
                    f"Acquire timed out after {timeout} seconds for amount={amount}"
                ) from error
        else:
            await self._semaphore_acquire(amount)

    @asynccontextmanager
    async def acquire_ctx(
        self, amount: float = 1, timeout: float | None = None
    ) -> AsyncGenerator[AsyncLeakyBucketGCRA]:
        """Async context manager that acquires a specific amount of capacity.

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

        Yields:
            AsyncGenerator[AsyncLeakyBucketGCRA]: The AsyncLeakyBucketGCRA instance
        """
        await self.acquire(amount=amount, timeout=timeout)
        try:
            yield self
        finally:
            pass

    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
        """
        diff = actual - estimated
        if diff > 0:
            await self.acquire(amount=diff)
        elif diff < 0:
            async with self._lock:
                t_a = time.monotonic()
                if self._last_leak is not None:
                    elapsed = t_a - self._last_leak
                    self._bucket_level = max(0.0, self._bucket_level - elapsed)
                    self._last_leak = t_a
                self._bucket_level = max(0.0, self._bucket_level + diff * self.T)

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

        Returns:
            An instance of the AsyncLeakyBucketGCRA class
        """
        await self.acquire()
        return self

    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
            exc_val: The value of the exception raised
            exc_tb: The traceback object
        """
        return None

__aenter__() async

Enter the context manager, acquiring resources if necessary

Returns:

Type Description
AsyncLeakyBucketGCRA

An instance of the AsyncLeakyBucketGCRA class
Source code in limitor/generic_cell_rate/core.py 
580
581
582
583
584
585
586
587
async def __aenter__(self) -> AsyncLeakyBucketGCRA:
    """Enter the context manager, acquiring resources if necessary

    Returns:
        An instance of the AsyncLeakyBucketGCRA class
    """
    await self.acquire()
    return self

__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

 required
exc_val BaseException | None

The value of the exception raised

 required
exc_tb TracebackType | None

The traceback object

 required
Source code in limitor/generic_cell_rate/core.py 
589
590
591
592
593
594
595
596
597
598
599
600
601
602
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
        exc_val: The value of the exception raised
        exc_tb: The traceback object
    """
    return None

acquire(amount=1, timeout=None) async

Acquire capacity, waiting asynchronously until allowed.

Supports timeout and cancellation.

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

Raises:

Type Description
TimeoutError

If the acquire operation times out after the specified timeout period
Source code in limitor/generic_cell_rate/core.py 
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
    """Acquire capacity, waiting asynchronously until allowed.

    Supports timeout and cancellation.

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

    Raises:
        TimeoutError: If the acquire operation times out after the specified timeout period
    """
    validate_amount(self, amount=amount)

    if timeout is not None:
        try:
            await asyncio.wait_for(self._semaphore_acquire(amount), timeout=timeout)
        except TimeoutError as error:
            raise TimeoutError(
                f"Acquire timed out after {timeout} seconds for amount={amount}"
            ) from error
    else:
        await self._semaphore_acquire(amount)

acquire_ctx(amount=1, timeout=None) async

Async context manager that acquires a specific 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

Yields:

Type Description
AsyncGenerator[AsyncLeakyBucketGCRA]

AsyncGenerator[AsyncLeakyBucketGCRA]: The AsyncLeakyBucketGCRA instance
Source code in limitor/generic_cell_rate/core.py 
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
@asynccontextmanager
async def acquire_ctx(
    self, amount: float = 1, timeout: float | None = None
) -> AsyncGenerator[AsyncLeakyBucketGCRA]:
    """Async context manager that acquires a specific amount of capacity.

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

    Yields:
        AsyncGenerator[AsyncLeakyBucketGCRA]: The AsyncLeakyBucketGCRA instance
    """
    await self.acquire(amount=amount, timeout=timeout)
    try:
        yield self
    finally:
        pass

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/generic_cell_rate/core.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
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
    """
    diff = actual - estimated
    if diff > 0:
        await self.acquire(amount=diff)
    elif diff < 0:
        async with self._lock:
            t_a = time.monotonic()
            if self._last_leak is not None:
                elapsed = t_a - self._last_leak
                self._bucket_level = max(0.0, self._bucket_level - elapsed)
                self._last_leak = t_a
            self._bucket_level = max(0.0, self._bucket_level + diff * self.T)

AsyncVirtualSchedulingGCRA

Virtual Scheduling Generic Cell Rate Algorithm Rate Limiter

Parameters:

Name Type Description Default
bucket_config BucketConfig | None

Configuration for the GCR algorithm with the max capacity and time period in seconds

 None
max_concurrent int | None

Maximum number of concurrent requests allowed to acquire capacity

 None
Note

This implementation is asynchronous and supports bursts up to the capacity within the specified time period

References

https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm

Source code in limitor/generic_cell_rate/core.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
class AsyncVirtualSchedulingGCRA:
    """Virtual Scheduling Generic Cell Rate Algorithm Rate Limiter

    Args:
        bucket_config: Configuration for the GCR algorithm with the max capacity and time period in seconds
        max_concurrent: Maximum number of concurrent requests allowed to acquire capacity

    Note:
        This implementation is asynchronous and supports bursts up to the capacity within the specified time period

    References:
        https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm
    """

    def __init__(
        self,
        bucket_config: BucketConfig | None = None,
        max_concurrent: int | None = None,
    ):
        # import config and set attributes
        config = bucket_config or BucketConfig()
        self.capacity = config.capacity
        self.seconds = config.seconds

        self.leak_rate = self.capacity / self.seconds  # units per second
        self.T = 1 / self.leak_rate  # time to leak one unit

        # burst rate, but can't do this if the amount is variable
        # self.tau = self.T * self.burst

        # theoretical arrival time (TAT)
        self._tat: float | None = None

        self.max_concurrent = max_concurrent
        self._lock = asyncio.Lock()
        self._semaphore: asyncio.Semaphore | AbstractAsyncContextManager[Any] | None = (
            None
        )

    async def _acquire_logic(self, amount: float = 1) -> None:
        """Acquire resources, blocking if necessary to conform to the rate limit

        Args:
            amount: The amount of capacity to check for, defaults to 1

        Notes:
            Adding a lock here ensures that the acquire logic is atomic, but it also means that the
                requests are going to be done in the order they were received  i.e. not out-of-order like
                most async programs.
            The benefit is that with multiple concurrent requests, we can ensure that the bucket level
                is updated correctly and that we don't have multiple requests trying to update the bucket level
                at the same time, which could lead to an inconsistent state i.e. a race condition.
        """
        async with (
            self._lock
        ):  # ensure atomicity given we can have multiple concurrent requests
            t_a = time.monotonic()
            if self._tat is None:
                # first cell
                self._tat = t_a

            # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
            tau = self.T * (self.capacity - amount)
            if t_a < self._tat - tau:
                delay = (self._tat - tau) - t_a
                await asyncio.sleep(delay)

            self._tat = max(t_a, self._tat) + amount * self.T

    async def _semaphore_acquire(self, amount: float = 1) -> None:
        """Acquire capacity using a semaphore to limit concurrency.

        Args:
            amount: The amount of capacity to acquire, defaults to 1
        """
        if self._semaphore is None:
            self._semaphore = (
                asyncio.Semaphore(self.max_concurrent)
                if self.max_concurrent
                else nullcontext()
            )

        async with self._semaphore:
            await self._acquire_logic(amount)

    async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
        """Acquire capacity, waiting asynchronously until allowed.

        Supports timeout and cancellation.

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

        Raises:
            TimeoutError: If the acquire operation times out after the specified timeout period
        """
        validate_amount(self, amount=amount)

        if timeout is not None:
            try:
                await asyncio.wait_for(self._semaphore_acquire(amount), timeout=timeout)
            except TimeoutError as error:
                raise TimeoutError(
                    f"Acquire timed out after {timeout} seconds for amount={amount}"
                ) from error
        else:
            await self._semaphore_acquire(amount)

    @asynccontextmanager
    async def acquire_ctx(
        self, amount: float = 1, timeout: float | None = None
    ) -> AsyncGenerator[AsyncVirtualSchedulingGCRA]:
        """Async context manager that acquires a specific amount of capacity.

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

        Yields:
            AsyncGenerator[AsyncVirtualSchedulingGCRA]: The AsyncVirtualSchedulingGCRA instance
        """
        await self.acquire(amount=amount, timeout=timeout)
        try:
            yield self
        finally:
            pass

    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
        """
        diff = actual - estimated
        if diff > 0:
            await self.acquire(amount=diff)
        elif diff < 0:
            async with self._lock:
                if self._tat is not None:
                    self._tat = max(time.monotonic(), self._tat + diff * self.T)

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

        Returns:
            An instance of the VirtualSchedulingGCRA class
        """
        await self.acquire()
        return self

    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
            exc_val: The value of the exception raised
            exc_tb: The traceback object
        """
        return None

__aenter__() async

Enter the context manager, acquiring resources if necessary

Returns:

Type Description
AsyncVirtualSchedulingGCRA

An instance of the VirtualSchedulingGCRA class
Source code in limitor/generic_cell_rate/core.py 
400
401
402
403
404
405
406
407
async def __aenter__(self) -> AsyncVirtualSchedulingGCRA:
    """Enter the context manager, acquiring resources if necessary

    Returns:
        An instance of the VirtualSchedulingGCRA class
    """
    await self.acquire()
    return self

__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

 required
exc_val BaseException | None

The value of the exception raised

 required
exc_tb TracebackType | None

The traceback object

 required
Source code in limitor/generic_cell_rate/core.py 
409
410
411
412
413
414
415
416
417
418
419
420
421
422
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
        exc_val: The value of the exception raised
        exc_tb: The traceback object
    """
    return None

acquire(amount=1, timeout=None) async

Acquire capacity, waiting asynchronously until allowed.

Supports timeout and cancellation.

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

Raises:

Type Description
TimeoutError

If the acquire operation times out after the specified timeout period
Source code in limitor/generic_cell_rate/core.py 
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
async def acquire(self, amount: float = 1, timeout: float | None = None) -> None:
    """Acquire capacity, waiting asynchronously until allowed.

    Supports timeout and cancellation.

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

    Raises:
        TimeoutError: If the acquire operation times out after the specified timeout period
    """
    validate_amount(self, amount=amount)

    if timeout is not None:
        try:
            await asyncio.wait_for(self._semaphore_acquire(amount), timeout=timeout)
        except TimeoutError as error:
            raise TimeoutError(
                f"Acquire timed out after {timeout} seconds for amount={amount}"
            ) from error
    else:
        await self._semaphore_acquire(amount)

acquire_ctx(amount=1, timeout=None) async

Async context manager that acquires a specific 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

Yields:

Type Description
AsyncGenerator[AsyncVirtualSchedulingGCRA]

AsyncGenerator[AsyncVirtualSchedulingGCRA]: The AsyncVirtualSchedulingGCRA instance
Source code in limitor/generic_cell_rate/core.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
@asynccontextmanager
async def acquire_ctx(
    self, amount: float = 1, timeout: float | None = None
) -> AsyncGenerator[AsyncVirtualSchedulingGCRA]:
    """Async context manager that acquires a specific amount of capacity.

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

    Yields:
        AsyncGenerator[AsyncVirtualSchedulingGCRA]: The AsyncVirtualSchedulingGCRA instance
    """
    await self.acquire(amount=amount, timeout=timeout)
    try:
        yield self
    finally:
        pass

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/generic_cell_rate/core.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
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
    """
    diff = actual - estimated
    if diff > 0:
        await self.acquire(amount=diff)
    elif diff < 0:
        async with self._lock:
            if self._tat is not None:
                self._tat = max(time.monotonic(), self._tat + diff * self.T)

SyncLeakyBucketGCRA

Continuous-state Leaky Bucket Rate Limiter

Parameters:

Name Type Description Default
bucket_config BucketConfig | None

Configuration for the GCR algorithm with the max capacity and time period in seconds

 None
Note

This implementation is synchronous and supports bursts up to the capacity within the specified time period

References

https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm

Source code in limitor/generic_cell_rate/core.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
class SyncLeakyBucketGCRA:
    """Continuous-state Leaky Bucket Rate Limiter

    Args:
        bucket_config: Configuration for the GCR algorithm with the max capacity and time period in seconds

    Note:
        This implementation is synchronous and supports bursts up to the capacity within the specified time period

    References:
        https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm
    """

    def __init__(self, bucket_config: BucketConfig | None = None):
        # import config and set attributes
        config = bucket_config or BucketConfig()
        self.capacity = config.capacity
        self.seconds = config.seconds

        self.leak_rate = self.capacity / self.seconds  # units per second
        self.T = 1 / self.leak_rate  # time to leak one unit

        # burst rate, but can't do this if the amount is variable
        # self.tau = self.T * self.burst

        self._bucket_level = 0.0  # current volume in the bucket
        self._last_leak: float | None = None  # same as last conforming time or LCT

        # thread-safe
        self._lock = threading.Lock()

    def acquire(self, amount: float = 1) -> None:
        """Acquire resources, blocking if necessary to conform to the rate limit

        Args:
            amount: The amount of resources to acquire (default is 1)
        """
        validate_amount(self, amount=amount)

        with self._lock:
            t_a = time.monotonic()
            if self._last_leak is None:
                # first cell
                self._bucket_level = 0
                self._last_leak = t_a

            elapsed = t_a - self._last_leak
            self._bucket_level = self._bucket_level - elapsed

            # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
            tau = self.T * (self.capacity - amount)
            if self._bucket_level > tau:
                delay = self._bucket_level - tau
                time.sleep(delay)

                self._bucket_level = self._bucket_level - delay
                t_a += delay

            self._bucket_level = max(0.0, self._bucket_level) + amount * self.T
            self._last_leak = t_a

    @contextmanager
    def acquire_ctx(self, amount: float = 1) -> Generator[SyncLeakyBucketGCRA]:
        """Context manager that acquires a specific amount of capacity.

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

        Yields:
            Generator[SyncLeakyBucketGCRA]: The SyncLeakyBucketGCRA instance
        """
        self.acquire(amount=amount)
        try:
            yield self
        finally:
            pass

    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
        """
        diff = actual - estimated
        if diff > 0:
            self.acquire(amount=diff)
        elif diff < 0:
            with self._lock:
                t_a = time.monotonic()
                if self._last_leak is not None:
                    elapsed = t_a - self._last_leak
                    self._bucket_level = max(0.0, self._bucket_level - elapsed)
                    self._last_leak = t_a
                self._bucket_level = max(0.0, self._bucket_level + diff * self.T)

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

        Returns:
            An instance of the LeakyBucketGCRA class
        """
        self.acquire()
        return self

    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
            exc_val: The value of the exception raised
            exc_tb: The traceback object
        """
        return None

__enter__()

Enter the context manager, acquiring resources if necessary

Returns:

Type Description
SyncLeakyBucketGCRA

An instance of the LeakyBucketGCRA class
Source code in limitor/generic_cell_rate/core.py 
232
233
234
235
236
237
238
239
def __enter__(self) -> SyncLeakyBucketGCRA:
    """Enter the context manager, acquiring resources if necessary

    Returns:
        An instance of the LeakyBucketGCRA class
    """
    self.acquire()
    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] | None

The type of the exception raised

 required
exc_val BaseException | None

The value of the exception raised

 required
exc_tb TracebackType | None

The traceback object

 required
Source code in limitor/generic_cell_rate/core.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
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
        exc_val: The value of the exception raised
        exc_tb: The traceback object
    """
    return None

acquire(amount=1)

Acquire resources, blocking if necessary to conform to the rate limit

Parameters:

Name Type Description Default
amount float

The amount of resources to acquire (default is 1)

 1
Source code in limitor/generic_cell_rate/core.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def acquire(self, amount: float = 1) -> None:
    """Acquire resources, blocking if necessary to conform to the rate limit

    Args:
        amount: The amount of resources to acquire (default is 1)
    """
    validate_amount(self, amount=amount)

    with self._lock:
        t_a = time.monotonic()
        if self._last_leak is None:
            # first cell
            self._bucket_level = 0
            self._last_leak = t_a

        elapsed = t_a - self._last_leak
        self._bucket_level = self._bucket_level - elapsed

        # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
        tau = self.T * (self.capacity - amount)
        if self._bucket_level > tau:
            delay = self._bucket_level - tau
            time.sleep(delay)

            self._bucket_level = self._bucket_level - delay
            t_a += delay

        self._bucket_level = max(0.0, self._bucket_level) + amount * self.T
        self._last_leak = t_a

acquire_ctx(amount=1)

Context manager that acquires a specific amount of capacity.

Parameters:

Name Type Description Default
amount float

The amount of capacity to acquire, defaults to 1

 1

Yields:

Type Description
Generator[SyncLeakyBucketGCRA]

Generator[SyncLeakyBucketGCRA]: The SyncLeakyBucketGCRA instance
Source code in limitor/generic_cell_rate/core.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
@contextmanager
def acquire_ctx(self, amount: float = 1) -> Generator[SyncLeakyBucketGCRA]:
    """Context manager that acquires a specific amount of capacity.

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

    Yields:
        Generator[SyncLeakyBucketGCRA]: The SyncLeakyBucketGCRA instance
    """
    self.acquire(amount=amount)
    try:
        yield self
    finally:
        pass

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/generic_cell_rate/core.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
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
    """
    diff = actual - estimated
    if diff > 0:
        self.acquire(amount=diff)
    elif diff < 0:
        with self._lock:
            t_a = time.monotonic()
            if self._last_leak is not None:
                elapsed = t_a - self._last_leak
                self._bucket_level = max(0.0, self._bucket_level - elapsed)
                self._last_leak = t_a
            self._bucket_level = max(0.0, self._bucket_level + diff * self.T)

SyncVirtualSchedulingGCRA

Virtual Scheduling Generic Cell Rate Algorithm Rate Limiter

Parameters:

Name Type Description Default
bucket_config BucketConfig | None

Configuration for the GCR algorithm with the max capacity and time period in seconds

 None
Note

This implementation is synchronous and supports bursts up to the capacity within the specified time period

References

https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm

Source code in limitor/generic_cell_rate/core.py 
 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
 77
 78
 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
class SyncVirtualSchedulingGCRA:
    """Virtual Scheduling Generic Cell Rate Algorithm Rate Limiter

    Args:
        bucket_config: Configuration for the GCR algorithm with the max capacity and time period in seconds

    Note:
        This implementation is synchronous and supports bursts up to the capacity within the specified time period

    References:
        https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm
    """

    def __init__(self, bucket_config: BucketConfig | None = None):
        # import config and set attributes
        config = bucket_config or BucketConfig()
        self.capacity = config.capacity
        self.seconds = config.seconds

        self.leak_rate = self.capacity / self.seconds  # units per second
        self.T = 1 / self.leak_rate  # time to leak one unit

        # burst rate, but can't do this if the amount is variable
        # self.tau = self.T * self.burst

        # theoretical arrival time (TAT)
        self._tat: float | None = None

        # thread-safe
        self._lock = threading.Lock()

    def acquire(self, amount: float = 1) -> None:
        """Acquire resources, blocking if necessary to conform to the rate limit

        Args:
            amount: The amount of resources to acquire (default is 1)
        """
        validate_amount(self, amount=amount)

        with self._lock:
            t_a = time.monotonic()
            if self._tat is None:
                # first cell
                self._tat = t_a

            # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
            tau = self.T * (self.capacity - amount)
            if t_a < self._tat - tau:
                delay = (self._tat - tau) - t_a
                time.sleep(delay)

            self._tat = max(t_a, self._tat) + amount * self.T

    @contextmanager
    def acquire_ctx(self, amount: float = 1) -> Generator[SyncVirtualSchedulingGCRA]:
        """Context manager that acquires a specific amount of capacity.

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

        Yields:
            Generator[SyncVirtualSchedulingGCRA]: The SyncVirtualSchedulingGCRA instance
        """
        self.acquire(amount=amount)
        try:
            yield self
        finally:
            pass

    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
        """
        diff = actual - estimated
        if diff > 0:
            self.acquire(amount=diff)
        elif diff < 0:
            with self._lock:
                if self._tat is not None:
                    self._tat = max(time.monotonic(), self._tat + diff * self.T)

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

        Returns:
            An instance of the VirtualSchedulingGCRA class
        """
        self.acquire()
        return self

    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
            exc_val: The value of the exception raised
            exc_tb: The traceback object
        """
        return None

__enter__()

Enter the context manager, acquiring resources if necessary

Returns:

Type Description
SyncVirtualSchedulingGCRA

An instance of the VirtualSchedulingGCRA class
Source code in limitor/generic_cell_rate/core.py 
111
112
113
114
115
116
117
118
def __enter__(self) -> SyncVirtualSchedulingGCRA:
    """Enter the context manager, acquiring resources if necessary

    Returns:
        An instance of the VirtualSchedulingGCRA class
    """
    self.acquire()
    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] | None

The type of the exception raised

 required
exc_val BaseException | None

The value of the exception raised

 required
exc_tb TracebackType | None

The traceback object

 required
Source code in limitor/generic_cell_rate/core.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
        exc_val: The value of the exception raised
        exc_tb: The traceback object
    """
    return None

acquire(amount=1)

Acquire resources, blocking if necessary to conform to the rate limit

Parameters:

Name Type Description Default
amount float

The amount of resources to acquire (default is 1)

 1
Source code in limitor/generic_cell_rate/core.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def acquire(self, amount: float = 1) -> None:
    """Acquire resources, blocking if necessary to conform to the rate limit

    Args:
        amount: The amount of resources to acquire (default is 1)
    """
    validate_amount(self, amount=amount)

    with self._lock:
        t_a = time.monotonic()
        if self._tat is None:
            # first cell
            self._tat = t_a

        # note: we can also make `self.capacity - amount` as class param = burst i.e. independent of capacity
        tau = self.T * (self.capacity - amount)
        if t_a < self._tat - tau:
            delay = (self._tat - tau) - t_a
            time.sleep(delay)

        self._tat = max(t_a, self._tat) + amount * self.T

acquire_ctx(amount=1)

Context manager that acquires a specific amount of capacity.

Parameters:

Name Type Description Default
amount float

The amount of capacity to acquire, defaults to 1

 1

Yields:

Type Description
Generator[SyncVirtualSchedulingGCRA]

Generator[SyncVirtualSchedulingGCRA]: The SyncVirtualSchedulingGCRA instance
Source code in limitor/generic_cell_rate/core.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@contextmanager
def acquire_ctx(self, amount: float = 1) -> Generator[SyncVirtualSchedulingGCRA]:
    """Context manager that acquires a specific amount of capacity.

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

    Yields:
        Generator[SyncVirtualSchedulingGCRA]: The SyncVirtualSchedulingGCRA instance
    """
    self.acquire(amount=amount)
    try:
        yield self
    finally:
        pass

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/generic_cell_rate/core.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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
    """
    diff = actual - estimated
    if diff > 0:
        self.acquire(amount=diff)
    elif diff < 0:
        with self._lock:
            if self._tat is not None:
                self._tat = max(time.monotonic(), self._tat + diff * self.T)