Source code for baseplate.lib.ratelimit.backends
import time
[docs]class RateLimitBackend:
"""An interface for rate limit backends to implement."""
[docs] def consume(self, key: str, amount: int, allowance: int, interval: int) -> bool:
"""Consume the given `amount` from the allowance for the given `key`.
This will return true if the `key` remains below the `allowance`
after consuming the given `amount`.
:param key: The name of the rate limit bucket to consume from.
:param amount: The amount to consume from the rate limit bucket.
:param allowance: The maximum allowance for the rate limit bucket.
:param interval: The interval to reset the allowance.
"""
raise NotImplementedError
def _get_current_bucket(bucket_size: int) -> str:
current_timestamp_seconds = int(time.time())
return str(current_timestamp_seconds // bucket_size)