API Reference

django_ratelimiter.decorator

ratelimit

ratelimit(
    rate: Union[str, Callable[[HttpRequest], str]],
    key: Union[str, Callable[[HttpRequest], str], None] = None,
    methods: Union[str, Sequence[str], None] = None,
    strategy: Literal[
        "fixed-window", "fixed-window-elastic-expiry", "moving-window"
    ] = "fixed-window",
    response: Optional[HttpResponse] = None,
    storage: Optional[Storage] = None,
    cache: Optional[str] = None,
) -> Callable[[ViewFunc], ViewFunc]

Rate limiting decorator for wrapping views.

Parameters:

Name	Type	Description	Default
rate	Union[str, Callable[[HttpRequest], str]]	rate string (i.e. 5/second) or a callable that takes a request and returns a rate	required
key	Union[str, Callable[[HttpRequest], str], None]	request attribute or callable that returns a string to be used as identifier	None
methods	Union[str, Sequence[str], None]	only rate limit specified method(s)	None
strategy	Literal['fixed-window', 'fixed-window-elastic-expiry', 'moving-window']	a name of rate limiting strategy	'fixed-window'
response	Optional[HttpResponse]	custom rate limit response instance	None
storage	Optional[Storage]	override default rate limit storage	None
cache	Optional[str]	override default cache name if using django cache storage backend	None

Source code in django_ratelimiter/decorator.py

def ratelimit(
    rate: Union[str, Callable[[HttpRequest], str]],
    key: Union[str, Callable[[HttpRequest], str], None] = None,
    methods: Union[str, Sequence[str], None] = None,
    strategy: Literal[
        "fixed-window",
        "fixed-window-elastic-expiry",
        "moving-window",
    ] = "fixed-window",
    response: Optional[HttpResponse] = None,
    storage: Optional[Storage] = None,
    cache: Optional[str] = None,
) -> Callable[[ViewFunc], ViewFunc]:
    """Rate limiting decorator for wrapping views.

    Arguments:
        rate: rate string (i.e. `5/second`) or a callable that takes a request and returns a rate
        key: request attribute or callable that returns a string to be used as identifier
        methods: only rate limit specified method(s)
        strategy: a name of rate limiting strategy
        response: custom rate limit response instance
        storage: override default rate limit storage
        cache: override default cache name if using django cache storage backend
    """
    if storage and cache:
        raise ValueError("Can't use both cache and storage")
    rate_limiter = get_rate_limiter(strategy, storage)

    def decorator(func: ViewFunc) -> ViewFunc:
        @wraps(func)
        def wrapper(
            request: HttpRequest, *args: P.args, **kwargs: P.kwargs
        ) -> HttpResponse:
            rate_str = rate(request) if callable(rate) else rate
            parsed_rate = parse(rate_str)
            if not methods or request.method in methods:
                identifiers = build_identifiers(func, methods)
                if key:
                    value = key(request) if callable(key) else getattr(request, key)
                    value = str(value.pk if isinstance(value, models.Model) else value)
                    identifiers.append(value)

                if not rate_limiter.hit(parsed_rate, *identifiers):
                    return response or HttpResponse(
                        "Too Many Requests",
                        status=429,
                    )
            return func(request, *args, **kwargs)

        return wrapper

    return decorator

django_ratelimiter.middleware

AbstractRateLimiterMiddleware

Bases: ABC

Abstract base class for rate limiting middleware.

Attributes:

Name	Type	Description
STRATEGY	str	default rate limiter strategy. Defaults to fixed-window.

Source code in django_ratelimiter/middleware.py

class AbstractRateLimiterMiddleware(abc.ABC):
    """Abstract base class for rate limiting middleware.

    Attributes:
        STRATEGY: default rate limiter strategy. Defaults to `fixed-window`.
    """

    STRATEGY: str = "fixed-window"

    def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
        self.get_response = get_response

    def storage_for(self, request: HttpRequest) -> Storage:
        """Override to set non-default storage."""
        return get_storage()

    def strategy_for(self, request: HttpRequest) -> str:
        """Override to customize strategy (i.e., based on a request path, method)"""
        return self.STRATEGY

    def keys_for(self, request: HttpRequest) -> list[str]:
        """By default, this will use middleware name for all requests,
        effectively this means global rate limiting for all requests.

        Override this method to rate-limit based on a request attribute like a path, user, etc.
        """
        return [f"{self.__class__.__module__}.{self.__class__.__qualname__}"]

    def ratelimit_response(self, request: HttpRequest) -> HttpResponse:
        """Override to return a custom response when rate limit is exceeded."""
        return HttpResponse("Too Many Requests", status=429)

    @abc.abstractmethod
    def rate_for(self, request: HttpRequest) -> Optional[str]:
        """Returns a rate for given request.

        If `None` is returned, request is not rate-limited.
        """

    def __call__(self, request: HttpRequest) -> HttpResponse:
        if rate := self.rate_for(request):
            parsed_rate = parse(rate)
            keys = self.keys_for(request)
            strategy = self.strategy_for(request)
            storage = self.storage_for(request)
            rate_limiter = get_rate_limiter(strategy, storage)
            if not rate_limiter.hit(parsed_rate, *keys):
                return self.ratelimit_response(request)
        return self.get_response(request)

keys_for

keys_for(request: HttpRequest) -> list[str]

By default, this will use middleware name for all requests, effectively this means global rate limiting for all requests.

Override this method to rate-limit based on a request attribute like a path, user, etc.

Source code in django_ratelimiter/middleware.py

def keys_for(self, request: HttpRequest) -> list[str]:
    """By default, this will use middleware name for all requests,
    effectively this means global rate limiting for all requests.

    Override this method to rate-limit based on a request attribute like a path, user, etc.
    """
    return [f"{self.__class__.__module__}.{self.__class__.__qualname__}"]

rate_for abstractmethod

rate_for(request: HttpRequest) -> Optional[str]

Returns a rate for given request.

If None is returned, request is not rate-limited.

Source code in django_ratelimiter/middleware.py

@abc.abstractmethod
def rate_for(self, request: HttpRequest) -> Optional[str]:
    """Returns a rate for given request.

    If `None` is returned, request is not rate-limited.
    """

ratelimit_response

ratelimit_response(request: HttpRequest) -> HttpResponse

Override to return a custom response when rate limit is exceeded.

Source code in django_ratelimiter/middleware.py

def ratelimit_response(self, request: HttpRequest) -> HttpResponse:
    """Override to return a custom response when rate limit is exceeded."""
    return HttpResponse("Too Many Requests", status=429)

storage_for

storage_for(request: HttpRequest) -> Storage

Override to set non-default storage.

Source code in django_ratelimiter/middleware.py

def storage_for(self, request: HttpRequest) -> Storage:
    """Override to set non-default storage."""
    return get_storage()

strategy_for

strategy_for(request: HttpRequest) -> str

Override to customize strategy (i.e., based on a request path, method)

Source code in django_ratelimiter/middleware.py

def strategy_for(self, request: HttpRequest) -> str:
    """Override to customize strategy (i.e., based on a request path, method)"""
    return self.STRATEGY

django_ratelimiter.storage

CacheStorage

Bases: Storage

Rate limiting storage with django cache backend.

Source code in django_ratelimiter/storage.py

class CacheStorage(Storage):
    """Rate limiting storage with django cache backend."""

    def __init__(
        self,
        cache: str,
        wrap_exceptions: bool = False,
        **options: Union[float, str, bool],
    ) -> None:
        self.cache: BaseCache = caches[cache]
        super().__init__(uri=None, wrap_exceptions=wrap_exceptions, **options)

    @property
    def base_exceptions(self) -> Union[type[Exception], tuple[type[Exception], ...]]:
        return Exception

    def get(self, key: str) -> int:
        return self.cache.get(key, 0)

    def incr(
        self, key: str, expiry: int, elastic_expiry: bool = False, amount: int = 1
    ) -> int:
        self.cache.get_or_set(key, 0, expiry)
        self.cache.get_or_set(f"{key}/expires", time.time() + expiry, expiry)
        try:
            value = self.cache.incr(key, amount) or amount
        except ValueError:
            value = amount
        if elastic_expiry:
            self.cache.touch(key, expiry)
            self.cache.set(f"{key}/expires", time.time() + expiry, expiry)
        return value

    def get_expiry(self, key: str) -> int:
        return int(float(self.cache.get(key + "/expires") or time.time()))

    def check(self) -> bool:
        try:
            self.cache.get("django-ratelimiter-check")
            return True
        except:  # noqa: E722
            return False

    def reset(self) -> Optional[int]:
        raise NotImplementedError

    def clear(self, key: str) -> None:
        self.cache.delete(key)

django_ratelimiter.utils

build_identifiers

build_identifiers(func: ViewFunc, methods: Union[str, Sequence[str], None] = None) -> list[str]

Build view identifiers for storage cache key using function signature and list of methods.

Source code in django_ratelimiter/utils.py

def build_identifiers(
    func: ViewFunc, methods: Union[str, Sequence[str], None] = None
) -> list[str]:
    """Build view identifiers for storage cache key using function signature and list of methods."""
    if isinstance(func, partial):
        # method_decorator scenario
        identifiers = [
            func.func.__self__.__class__.__module__,
            f"{func.func.__self__.__class__.__qualname__}.{func.func.__name__}",
        ]
    else:
        identifiers = [func.__module__, func.__qualname__]
    if methods:
        methods_ = methods if isinstance(methods, str) else "|".join(sorted(methods))
        identifiers.append(methods_)
    return identifiers

get_rate_limiter

get_rate_limiter(strategy: str, storage: Optional[Storage] = None) -> RateLimiter

Return a ratelimiter instance for given strategy.

Source code in django_ratelimiter/utils.py

def get_rate_limiter(strategy: str, storage: Optional[Storage] = None) -> RateLimiter:
    """Return a ratelimiter instance for given strategy."""
    if strategy not in STRATEGIES:
        raise ValueError(
            f"Unknown strategy {strategy}, must be one of {STRATEGIES.keys()}"
        )
    storage = storage or get_storage()
    return STRATEGIES[strategy](storage)

get_storage cached

get_storage() -> Storage

Returns a default storage backend instance, defined by either DJANGO_RATELIMITER_CACHE or DJANGO_RATELIMITER_STORAGE.

Source code in django_ratelimiter/utils.py

@lru_cache(maxsize=None)
def get_storage() -> Storage:
    """Returns a default storage backend instance, defined by either `DJANGO_RATELIMITER_CACHE`
    or `DJANGO_RATELIMITER_STORAGE`."""
    cache_name: Optional[str] = getattr(settings, "DJANGO_RATELIMITER_CACHE", None)
    storage: Optional[Storage] = getattr(settings, "DJANGO_RATELIMITER_STORAGE", None)
    if cache_name and storage:
        raise ValueError(
            "DJANGO_RATELIMITER_CACHE and DJANGO_RATELIMITER_STORAGE can't be used together"
        )
    return storage or CacheStorage(cache_name or "default")

django_ratelimiter.types.P module-attribute

P = ParamSpec('P')

django_ratelimiter.types.ViewFunc module-attribute

ViewFunc = Callable[Concatenate[HttpRequest, P], HttpResponse]