Source code for

import threading
from abc import ABC, abstractmethod

from import StorageRegistry
from limits.typing import List, Optional, Tuple, Union
from limits.util import LazyDependency

[docs] class Storage(LazyDependency, metaclass=StorageRegistry): """ Base class to extend when implementing a storage backend. """ STORAGE_SCHEME: Optional[List[str]] """The storage schemes to register against this implementation""" def __init__(self, uri: Optional[str] = None, **options: Union[float, str, bool]): self.lock = threading.RLock() super().__init__()
[docs] @abstractmethod def incr( self, key: str, expiry: int, elastic_expiry: bool = False, amount: int = 1 ) -> int: """ increments the counter for a given rate limit key :param key: the key to increment :param expiry: amount in seconds for the key to expire in :param elastic_expiry: whether to keep extending the rate limit window every hit. :param amount: the number to increment by """ raise NotImplementedError
[docs] @abstractmethod def get(self, key: str) -> int: """ :param key: the key to get the counter value for """ raise NotImplementedError
[docs] @abstractmethod def get_expiry(self, key: str) -> int: """ :param key: the key to get the expiry for """ raise NotImplementedError
[docs] @abstractmethod def check(self) -> bool: """ check if storage is healthy """ raise NotImplementedError
[docs] @abstractmethod def reset(self) -> Optional[int]: """ reset storage to clear limits """ raise NotImplementedError
[docs] @abstractmethod def clear(self, key: str) -> None: """ resets the rate limit key :param key: the key to clear rate limits for """ raise NotImplementedError
[docs] class MovingWindowSupport(ABC): """ Abstract base for storages that intend to support the moving window strategy """
[docs] def acquire_entry(self, key: str, limit: int, expiry: int, amount: int = 1) -> bool: """ :param key: rate limit key to acquire an entry in :param limit: amount of entries allowed :param expiry: expiry of the entry :param amount: the number of entries to acquire """ raise NotImplementedError
[docs] def get_moving_window(self, key: str, limit: int, expiry: int) -> Tuple[int, int]: """ returns the starting point and the number of entries in the moving window :param key: rate limit key :param expiry: expiry of entry :return: (start of window, number of acquired entries) """ raise NotImplementedError