API Reference

Generated in part from source docstrings via mkdocstrings.

Throttles

rate_limit

Build a :class:RateLimitThrottle subclass for throttle_classes.

Parameters:

Name	Type	Description	Default
rate	RateSpec	See :attr:RateLimitThrottle.rate.	'60/m'
algorithm	str	One of "fixed_window", "sliding_window", "token_bucket" (default).	'token_bucket'
key	str | KeyFunc	See :attr:RateLimitThrottle.key.	'ip'
cost	int | Callable[[Any], int]	See :attr:RateLimitThrottle.cost.	1
burst	int | None	See :attr:RateLimitThrottle.burst.	None
scope	str | None	See :attr:RateLimitThrottle.scope.	None
tier_resolver	TierResolver	See :attr:RateLimitThrottle.tier_resolver.	default_tier_resolver
client	Any	See :attr:RateLimitThrottle.client.	None

Returns:

Type	Description
type[RateLimitThrottle]	A new :class:RateLimitThrottle subclass, ready to use in
type[RateLimitThrottle]	throttle_classes = [rate_limit(...)].

Raises:

Type	Description
InvalidAlgorithmError	If algorithm isn't recognized.

Example

.. code-block:: python

class ArticleViewSet(viewsets.ModelViewSet):
    throttle_classes = [
        rate_limit(rate="100/m", algorithm="token_bucket", burst=20, key="user"),
    ]

Source code in src/drf_ratelimit_plus/throttles.py

def rate_limit(
    *,
    rate: RateSpec = "60/m",
    algorithm: str = "token_bucket",
    key: str | KeyFunc = "ip",
    cost: int | Callable[[Any], int] = 1,
    burst: int | None = None,
    scope: str | None = None,
    tier_resolver: TierResolver = default_tier_resolver,
    client: Any = None,
) -> type[RateLimitThrottle]:
    """Build a :class:`RateLimitThrottle` subclass for ``throttle_classes``.

    Args:
        rate: See :attr:`RateLimitThrottle.rate`.
        algorithm: One of ``"fixed_window"``, ``"sliding_window"``,
            ``"token_bucket"`` (default).
        key: See :attr:`RateLimitThrottle.key`.
        cost: See :attr:`RateLimitThrottle.cost`.
        burst: See :attr:`RateLimitThrottle.burst`.
        scope: See :attr:`RateLimitThrottle.scope`.
        tier_resolver: See :attr:`RateLimitThrottle.tier_resolver`.
        client: See :attr:`RateLimitThrottle.client`.

    Returns:
        A new :class:`RateLimitThrottle` subclass, ready to use in
        ``throttle_classes = [rate_limit(...)]``.

    Raises:
        drf_ratelimit_plus.exceptions.InvalidAlgorithmError: If
            ``algorithm`` isn't recognized.

    Example:
        .. code-block:: python

            class ArticleViewSet(viewsets.ModelViewSet):
                throttle_classes = [
                    rate_limit(rate="100/m", algorithm="token_bucket", burst=20, key="user"),
                ]
    """
    if algorithm not in _ALGORITHM_NAMES:
        raise InvalidAlgorithmError(
            f"Unknown algorithm {algorithm!r}. Valid options: {_ALGORITHM_NAMES}."
        )
    return type(
        "ConfiguredRateLimitThrottle",
        (RateLimitThrottle,),
        {
            "algorithm": algorithm,
            "rate": staticmethod(rate) if callable(rate) else rate,
            "key": staticmethod(key) if callable(key) else key,
            "cost": staticmethod(cost) if callable(cost) else cost,
            "burst": burst,
            "scope": scope,
            "tier_resolver": staticmethod(tier_resolver),
            "client": client,
        },
    )

RateLimitThrottle

Bases: BaseThrottle

A configurable rate-limiting throttle.

Don't instantiate or subclass this directly for one-off configuration — use :func:rate_limit instead, which builds a properly configured subclass. Subclass this directly only if you want a reusable named throttle class (the same pattern DRF's own UserRateThrottle uses).

Attributes:

Name	Type	Description
algorithm	str	One of "fixed_window", "sliding_window", "token_bucket".
rate	RateSpec	A rate string ("100/m"), an already-parsed :class:~drf_ratelimit_plus.rates.Rate, a mapping of tier name to rate (for plan-tier support), or a callable taking the request and returning any of the above (for fully dynamic, e.g. database-driven, configuration).
key	str | KeyFunc	A built-in key function name ("ip", "user", "api_key", "tenant", "view") or a custom callable. See :mod:drf_ratelimit_plus.keys.
cost	int | Callable[[Any], int]	How many units a single request consumes — an int, or a callable taking the request and returning one, for per-request weighted costs.
burst	int | None	Token-bucket capacity (ignored by the other algorithms). Defaults to the rate's count when None.
scope	str | None	An explicit scope string grouping requests into the same limit regardless of which view handles them. Defaults to the handling view's dotted class path (i.e. each view gets an independent limit by default).
tier_resolver	TierResolver	Determines which tier name applies when rate is a mapping. Defaults to :func:~drf_ratelimit_plus.tiers.default_tier_resolver.
client	Any	An explicit Redis client to use instead of the one resolved from settings (see :func:~drf_ratelimit_plus.client.get_redis_client).

Note

If you subclass this directly (rather than using :func:rate_limit) and want rate, key, or cost to be a plain function rather than a string/int, wrap it in staticmethod(...) explicitly — otherwise Python's normal attribute lookup binds it as an instance method (implicitly passing self as the first argument), which isn't what these callables expect. :func:rate_limit handles this wrapping for you automatically.

Source code in src/drf_ratelimit_plus/throttles.py

class RateLimitThrottle(BaseThrottle):
    """A configurable rate-limiting throttle.

    Don't instantiate or subclass this directly for one-off configuration
    — use :func:`rate_limit` instead, which builds a properly configured
    subclass. Subclass this directly only if you want a *reusable* named
    throttle class (the same pattern DRF's own ``UserRateThrottle`` uses).

    Attributes:
        algorithm: One of ``"fixed_window"``, ``"sliding_window"``,
            ``"token_bucket"``.
        rate: A rate string (``"100/m"``), an already-parsed
            :class:`~drf_ratelimit_plus.rates.Rate`, a mapping of tier
            name to rate (for plan-tier support), or a callable taking
            the request and returning any of the above (for fully
            dynamic, e.g. database-driven, configuration).
        key: A built-in key function name (``"ip"``, ``"user"``,
            ``"api_key"``, ``"tenant"``, ``"view"``) or a custom
            callable. See :mod:`drf_ratelimit_plus.keys`.
        cost: How many units a single request consumes — an int, or a
            callable taking the request and returning one, for
            per-request weighted costs.
        burst: Token-bucket capacity (ignored by the other algorithms).
            Defaults to the rate's count when ``None``.
        scope: An explicit scope string grouping requests into the same
            limit regardless of which view handles them. Defaults to the
            handling view's dotted class path (i.e. each view gets an
            independent limit by default).
        tier_resolver: Determines which tier name applies when ``rate``
            is a mapping. Defaults to
            :func:`~drf_ratelimit_plus.tiers.default_tier_resolver`.
        client: An explicit Redis client to use instead of the one
            resolved from settings (see
            :func:`~drf_ratelimit_plus.client.get_redis_client`).

    Note:
        If you subclass this directly (rather than using
        :func:`rate_limit`) and want ``rate``, ``key``, or ``cost`` to be
        a plain function rather than a string/int, wrap it in
        ``staticmethod(...)`` explicitly — otherwise Python's normal
        attribute lookup binds it as an instance method (implicitly
        passing ``self`` as the first argument), which isn't what these
        callables expect. :func:`rate_limit` handles this wrapping for
        you automatically.
    """

    algorithm: str = "token_bucket"
    rate: RateSpec = "60/m"
    key: str | KeyFunc = "ip"
    cost: int | Callable[[Any], int] = 1
    burst: int | None = None
    scope: str | None = None
    tier_resolver: TierResolver = staticmethod(default_tier_resolver)
    client: Any = None

    def __init__(self) -> None:
        self.last_result: LimitResult | None = None

    def allow_request(self, request: Any, view: Any) -> bool:
        """See :meth:`rest_framework.throttling.BaseThrottle.allow_request`.

        Args:
            request: The incoming request.
            view: The view handling it.

        Returns:
            Whether the request is within its configured limit. The full
            :class:`~drf_ratelimit_plus.results.LimitResult` is stored on
            ``self.last_result`` regardless, for header construction by
            :class:`~drf_ratelimit_plus.mixins.RateLimitHeadersMixin`.
        """
        rate = self._resolve_rate(request)
        cost = self.cost(request) if callable(self.cost) else self.cost
        client = self.client if self.client is not None else get_redis_client()
        key = self._build_key(request, view)

        if self.algorithm == "token_bucket":
            result = token_bucket_check(client, key, rate, cost=cost, burst=self.burst)
        elif self.algorithm == "sliding_window":
            result = sliding_window_check(client, key, rate, cost=cost)
        elif self.algorithm == "fixed_window":
            result = fixed_window_check(client, key, rate, cost=cost)
        else:
            raise InvalidAlgorithmError(
                f"Unknown algorithm {self.algorithm!r}. Valid options: {_ALGORITHM_NAMES}."
            )

        self.last_result = result
        return result.allowed

    def wait(self) -> float | None:
        """See :meth:`rest_framework.throttling.BaseThrottle.wait`.

        Returns:
            Seconds the client should wait before retrying, or ``None``
            if the last check succeeded (or hasn't run yet).
        """
        if self.last_result is None:
            return None
        return self.last_result.retry_after

    def _build_key(self, request: Any, view: Any) -> str:
        key_func = resolve_key_func(self.key)
        scope = self.scope if self.scope is not None else by_view(request, view)
        identity = key_func(request, view)
        prefix = get_setting("KEY_PREFIX")
        return f"{prefix}{self.algorithm}:{scope}:{identity}"

    def _resolve_rate(self, request: Any) -> Rate:
        rate = self.rate
        if callable(rate):
            rate = rate(request)
        if isinstance(rate, Mapping):
            return resolve_tier_rate(request, rate, resolver=self.tier_resolver)
        return parse_rate(rate)

allow_request(request, view)

See :meth:rest_framework.throttling.BaseThrottle.allow_request.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	The view handling it.	required

Returns:

Type	Description
bool	Whether the request is within its configured limit. The full
bool	class:~drf_ratelimit_plus.results.LimitResult is stored on
bool	self.last_result regardless, for header construction by
bool	class:~drf_ratelimit_plus.mixins.RateLimitHeadersMixin.

Source code in src/drf_ratelimit_plus/throttles.py

def allow_request(self, request: Any, view: Any) -> bool:
    """See :meth:`rest_framework.throttling.BaseThrottle.allow_request`.

    Args:
        request: The incoming request.
        view: The view handling it.

    Returns:
        Whether the request is within its configured limit. The full
        :class:`~drf_ratelimit_plus.results.LimitResult` is stored on
        ``self.last_result`` regardless, for header construction by
        :class:`~drf_ratelimit_plus.mixins.RateLimitHeadersMixin`.
    """
    rate = self._resolve_rate(request)
    cost = self.cost(request) if callable(self.cost) else self.cost
    client = self.client if self.client is not None else get_redis_client()
    key = self._build_key(request, view)

    if self.algorithm == "token_bucket":
        result = token_bucket_check(client, key, rate, cost=cost, burst=self.burst)
    elif self.algorithm == "sliding_window":
        result = sliding_window_check(client, key, rate, cost=cost)
    elif self.algorithm == "fixed_window":
        result = fixed_window_check(client, key, rate, cost=cost)
    else:
        raise InvalidAlgorithmError(
            f"Unknown algorithm {self.algorithm!r}. Valid options: {_ALGORITHM_NAMES}."
        )

    self.last_result = result
    return result.allowed

wait()

See :meth:rest_framework.throttling.BaseThrottle.wait.

Returns:

Type	Description
float | None	Seconds the client should wait before retrying, or None
float | None	if the last check succeeded (or hasn't run yet).

Source code in src/drf_ratelimit_plus/throttles.py

def wait(self) -> float | None:
    """See :meth:`rest_framework.throttling.BaseThrottle.wait`.

    Returns:
        Seconds the client should wait before retrying, or ``None``
        if the last check succeeded (or hasn't run yet).
    """
    if self.last_result is None:
        return None
    return self.last_result.retry_after

Decorator

ratelimit

Apply rate limiting to a single function-based view.

Parameters:

Name	Type	Description	Default
rate	RateSpec	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.rate.	'60/m'
algorithm	str	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.algorithm.	'token_bucket'
key	str | KeyFunc	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.key.	'ip'
cost	int | Callable[[Any], int]	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.cost.	1
burst	int | None	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.burst.	None
scope	str | None	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.scope.	None
tier_resolver	TierResolver	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.tier_resolver.	default_tier_resolver
client	Any	See :attr:~drf_ratelimit_plus.throttles.RateLimitThrottle.client.	None

Returns:

Type	Description
Callable[[_F], _F]	A decorator that enforces the configured limit, raising
Callable[[_F], _F]	class:rest_framework.exceptions.Throttled (which DRF converts
Callable[[_F], _F]	to 429 with a Retry-After header) when exceeded, and
Callable[[_F], _F]	adding RateLimit-* headers to successful responses.

Example

.. code-block:: python

from rest_framework.decorators import api_view
from drf_ratelimit_plus import ratelimit


@api_view(["POST"])
@ratelimit(rate="10/m", key="user")
def expensive_action(request):
    ...

Source code in src/drf_ratelimit_plus/decorators.py

def ratelimit(
    *,
    rate: RateSpec = "60/m",
    algorithm: str = "token_bucket",
    key: str | KeyFunc = "ip",
    cost: int | Callable[[Any], int] = 1,
    burst: int | None = None,
    scope: str | None = None,
    tier_resolver: TierResolver = default_tier_resolver,
    client: Any = None,
) -> Callable[[_F], _F]:
    """Apply rate limiting to a single function-based view.

    Args:
        rate: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.rate`.
        algorithm: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.algorithm`.
        key: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.key`.
        cost: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.cost`.
        burst: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.burst`.
        scope: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.scope`.
        tier_resolver: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.tier_resolver`.
        client: See :attr:`~drf_ratelimit_plus.throttles.RateLimitThrottle.client`.

    Returns:
        A decorator that enforces the configured limit, raising
        :class:`rest_framework.exceptions.Throttled` (which DRF converts
        to ``429`` with a ``Retry-After`` header) when exceeded, and
        adding ``RateLimit-*`` headers to successful responses.

    Example:
        .. code-block:: python

            from rest_framework.decorators import api_view
            from drf_ratelimit_plus import ratelimit


            @api_view(["POST"])
            @ratelimit(rate="10/m", key="user")
            def expensive_action(request):
                ...
    """
    throttle_class = rate_limit(
        rate=rate,
        algorithm=algorithm,
        key=key,
        cost=cost,
        burst=burst,
        scope=scope,
        tier_resolver=tier_resolver,
        client=client,
    )

    def decorator(view_func: _F) -> _F:
        @wraps(view_func)
        def wrapper(request: Any, *args: Any, **kwargs: Any) -> Any:
            throttle = throttle_class()
            if not throttle.allow_request(request, None):
                raise Throttled(wait=throttle.wait())
            response = view_func(request, *args, **kwargs)
            result = throttle.last_result
            if result is not None:
                response[get_setting("LIMIT_HEADER")] = str(result.limit)
                response[get_setting("REMAINING_HEADER")] = str(result.remaining)
                response[get_setting("RESET_HEADER")] = str(result.reset_seconds)
            return response

        return wrapper  # type: ignore[return-value]

    return decorator

Headers

RateLimitHeadersMixin

Adds RateLimit-* headers to every response, based on the :class:~drf_ratelimit_plus.throttles.RateLimitThrottle instances checked for the current request.

Mix into any APIView/GenericAPIView/viewset that uses :func:~drf_ratelimit_plus.throttles.rate_limit in its throttle_classes:

.. code-block:: python

class ArticleViewSet(RateLimitHeadersMixin, viewsets.ModelViewSet):
    throttle_classes = [rate_limit(rate="100/m")]

Must come before the DRF base class in the MRO, same as any other DRF mixin.

Source code in src/drf_ratelimit_plus/mixins.py

class RateLimitHeadersMixin:
    """Adds ``RateLimit-*`` headers to every response, based on the
    :class:`~drf_ratelimit_plus.throttles.RateLimitThrottle` instances
    checked for the current request.

    Mix into any ``APIView``/``GenericAPIView``/viewset that uses
    :func:`~drf_ratelimit_plus.throttles.rate_limit` in its
    ``throttle_classes``:

    .. code-block:: python

        class ArticleViewSet(RateLimitHeadersMixin, viewsets.ModelViewSet):
            throttle_classes = [rate_limit(rate="100/m")]

    Must come before the DRF base class in the MRO, same as any other
    DRF mixin.
    """

    def get_throttles(self) -> list[Any]:
        """Return (and cache) this request's throttle instances.

        DRF's own ``get_throttles()`` constructs a fresh instance from
        ``throttle_classes`` on every call — calling it more than once
        per request would mean ``check_throttles()`` and
        ``finalize_response()`` each see *different* throttle instances,
        so the ``last_result`` this mixin reads for headers would never
        match what actually decided the request's outcome. Caching on
        ``self`` fixes this: a view instance handles exactly one request
        in DRF, so caching per-instance is exactly "per request".

        Returns:
            The same list of throttle instances for every call during
            this request.
        """
        if not hasattr(self, "_ratelimit_plus_throttles"):
            self._ratelimit_plus_throttles = super().get_throttles()  # type: ignore[misc]
        throttles: list[Any] = self._ratelimit_plus_throttles
        return throttles

    def finalize_response(self, request: Any, response: Any, *args: Any, **kwargs: Any) -> Any:
        """Add ``RateLimit-*`` headers before returning the response.

        Args:
            request: The current request.
            response: The response produced by the view (or by DRF's
                exception handling, for a throttled request).
            *args: Passed through to the next class in the MRO.
            **kwargs: Passed through to the next class in the MRO.

        Returns:
            The response, with ``RateLimit-Limit``, ``RateLimit-Remaining``,
            and ``RateLimit-Reset`` headers added if any configured
            :class:`~drf_ratelimit_plus.throttles.RateLimitThrottle`
            produced a result for this request. If multiple such
            throttles were checked, the one with the *lowest* remaining
            count is reported (the one closest to being exceeded is the
            most informative to the client).
        """
        response = super().finalize_response(request, response, *args, **kwargs)  # type: ignore[misc]
        results = [
            t.last_result
            for t in self.get_throttles()
            if isinstance(t, RateLimitThrottle) and t.last_result is not None
        ]
        if not results:
            return response
        tightest = min(results, key=lambda r: r.remaining)
        response[get_setting("LIMIT_HEADER")] = str(tightest.limit)
        response[get_setting("REMAINING_HEADER")] = str(tightest.remaining)
        response[get_setting("RESET_HEADER")] = str(tightest.reset_seconds)
        return response

finalize_response(request, response, *args, **kwargs)

Add RateLimit-* headers before returning the response.

Parameters:

Name	Type	Description	Default
request	Any	The current request.	required
response	Any	The response produced by the view (or by DRF's exception handling, for a throttled request).	required
*args	Any	Passed through to the next class in the MRO.	()
**kwargs	Any	Passed through to the next class in the MRO.	{}

Returns:

Type	Description
Any	The response, with RateLimit-Limit, RateLimit-Remaining,
Any	and RateLimit-Reset headers added if any configured
Any	class:~drf_ratelimit_plus.throttles.RateLimitThrottle
Any	produced a result for this request. If multiple such
Any	throttles were checked, the one with the lowest remaining
Any	count is reported (the one closest to being exceeded is the
Any	most informative to the client).

Source code in src/drf_ratelimit_plus/mixins.py

def finalize_response(self, request: Any, response: Any, *args: Any, **kwargs: Any) -> Any:
    """Add ``RateLimit-*`` headers before returning the response.

    Args:
        request: The current request.
        response: The response produced by the view (or by DRF's
            exception handling, for a throttled request).
        *args: Passed through to the next class in the MRO.
        **kwargs: Passed through to the next class in the MRO.

    Returns:
        The response, with ``RateLimit-Limit``, ``RateLimit-Remaining``,
        and ``RateLimit-Reset`` headers added if any configured
        :class:`~drf_ratelimit_plus.throttles.RateLimitThrottle`
        produced a result for this request. If multiple such
        throttles were checked, the one with the *lowest* remaining
        count is reported (the one closest to being exceeded is the
        most informative to the client).
    """
    response = super().finalize_response(request, response, *args, **kwargs)  # type: ignore[misc]
    results = [
        t.last_result
        for t in self.get_throttles()
        if isinstance(t, RateLimitThrottle) and t.last_result is not None
    ]
    if not results:
        return response
    tightest = min(results, key=lambda r: r.remaining)
    response[get_setting("LIMIT_HEADER")] = str(tightest.limit)
    response[get_setting("REMAINING_HEADER")] = str(tightest.remaining)
    response[get_setting("RESET_HEADER")] = str(tightest.reset_seconds)
    return response

get_throttles()

Return (and cache) this request's throttle instances.

DRF's own get_throttles() constructs a fresh instance from throttle_classes on every call — calling it more than once per request would mean check_throttles() and finalize_response() each see different throttle instances, so the last_result this mixin reads for headers would never match what actually decided the request's outcome. Caching on self fixes this: a view instance handles exactly one request in DRF, so caching per-instance is exactly "per request".

Returns:

Type	Description
list[Any]	The same list of throttle instances for every call during
list[Any]	this request.

Source code in src/drf_ratelimit_plus/mixins.py

def get_throttles(self) -> list[Any]:
    """Return (and cache) this request's throttle instances.

    DRF's own ``get_throttles()`` constructs a fresh instance from
    ``throttle_classes`` on every call — calling it more than once
    per request would mean ``check_throttles()`` and
    ``finalize_response()`` each see *different* throttle instances,
    so the ``last_result`` this mixin reads for headers would never
    match what actually decided the request's outcome. Caching on
    ``self`` fixes this: a view instance handles exactly one request
    in DRF, so caching per-instance is exactly "per request".

    Returns:
        The same list of throttle instances for every call during
        this request.
    """
    if not hasattr(self, "_ratelimit_plus_throttles"):
        self._ratelimit_plus_throttles = super().get_throttles()  # type: ignore[misc]
    throttles: list[Any] = self._ratelimit_plus_throttles
    return throttles

Keys

by_ip

Key by the client's IP address.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	Unused; accepted for calling-convention consistency.	None

Returns:

Type	Description
str	X-Forwarded-For's first entry if present (the original client
str	in a proxied setup), otherwise REMOTE_ADDR.

Source code in src/drf_ratelimit_plus/keys.py

def by_ip(request: Any, view: Any = None) -> str:
    """Key by the client's IP address.

    Args:
        request: The incoming request.
        view: Unused; accepted for calling-convention consistency.

    Returns:
        ``X-Forwarded-For``'s first entry if present (the original client
        in a proxied setup), otherwise ``REMOTE_ADDR``.
    """
    forwarded = request.META.get("HTTP_X_FORWARDED_FOR")
    if forwarded:
        return str(forwarded).split(",")[0].strip()
    return str(request.META.get("REMOTE_ADDR", "unknown"))

by_user

Key by the authenticated user, falling back to IP for anonymous requests.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	Unused; accepted for calling-convention consistency.	None

Returns:

Name	Type	Description
	str	f"user:{pk}" for an authenticated user, otherwise the same
value	str	func:by_ip would produce.

Source code in src/drf_ratelimit_plus/keys.py

def by_user(request: Any, view: Any = None) -> str:
    """Key by the authenticated user, falling back to IP for anonymous requests.

    Args:
        request: The incoming request.
        view: Unused; accepted for calling-convention consistency.

    Returns:
        ``f"user:{pk}"`` for an authenticated user, otherwise the same
        value :func:`by_ip` would produce.
    """
    user = getattr(request, "user", None)
    if user is not None and getattr(user, "is_authenticated", False):
        return f"user:{user.pk}"
    return by_ip(request)

by_api_key

Key by an API key header, falling back to IP if absent.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	Unused; accepted for calling-convention consistency.	None

Returns:

Type	Description
str	f"apikey:{value}" using the header named by the
str	API_KEY_HEADER setting, or the same value :func:by_ip would
str	produce if the header is missing.

Source code in src/drf_ratelimit_plus/keys.py

def by_api_key(request: Any, view: Any = None) -> str:
    """Key by an API key header, falling back to IP if absent.

    Args:
        request: The incoming request.
        view: Unused; accepted for calling-convention consistency.

    Returns:
        ``f"apikey:{value}"`` using the header named by the
        ``API_KEY_HEADER`` setting, or the same value :func:`by_ip` would
        produce if the header is missing.
    """
    header = get_setting("API_KEY_HEADER")
    value = request.headers.get(header)
    if value:
        return f"apikey:{value}"
    return by_ip(request)

by_tenant

Key by a tenant identifier attached to the request, falling back to IP.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	Unused; accepted for calling-convention consistency.	None

Returns:

Type	Description
str	f"tenant:{value}" using the attribute named by the
str	TENANT_ATTR setting, or the same value :func:by_ip would
str	produce if that attribute is absent or None.

Source code in src/drf_ratelimit_plus/keys.py

def by_tenant(request: Any, view: Any = None) -> str:
    """Key by a tenant identifier attached to the request, falling back to IP.

    Args:
        request: The incoming request.
        view: Unused; accepted for calling-convention consistency.

    Returns:
        ``f"tenant:{value}"`` using the attribute named by the
        ``TENANT_ATTR`` setting, or the same value :func:`by_ip` would
        produce if that attribute is absent or ``None``.
    """
    attr = get_setting("TENANT_ATTR")
    value = getattr(request, attr, None)
    if value is not None:
        return f"tenant:{value}"
    return by_ip(request)

by_view

Key by the view/endpoint handling the request.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
view	Any	The view instance handling it, if available.	None

Returns:

Type	Description
str	A dotted path identifying the view's class, or the request path
str	if no view instance is available (e.g. a function-based view
str	calling this directly).

Source code in src/drf_ratelimit_plus/keys.py

def by_view(request: Any, view: Any = None) -> str:
    """Key by the view/endpoint handling the request.

    Args:
        request: The incoming request.
        view: The view instance handling it, if available.

    Returns:
        A dotted path identifying the view's class, or the request path
        if no view instance is available (e.g. a function-based view
        calling this directly).
    """
    if view is not None:
        cls = type(view)
        return f"view:{cls.__module__}.{cls.__qualname__}"
    return f"path:{request.path}"

combine

Combine multiple key functions into one, joined by |.

Useful for compound scoping, e.g. per-tenant-per-user: combine("tenant", "user").

Parameters:

Name	Type	Description	Default
*key_funcs	str | KeyFunc	Key specifications (names or callables), applied in order and joined with |.	()

Returns:

Type	Description
KeyFunc	A single key function producing the joined result.

Example

from unittest.mock import Mock request = Mock(user=Mock(is_authenticated=True, pk=42), META={}, headers={}) request.tenant_id = "acme" combine("tenant", "user")(request) 'tenant:acme|user:42'

Source code in src/drf_ratelimit_plus/keys.py

def combine(*key_funcs: str | KeyFunc) -> KeyFunc:
    """Combine multiple key functions into one, joined by ``|``.

    Useful for compound scoping, e.g. per-tenant-per-user:
    ``combine("tenant", "user")``.

    Args:
        *key_funcs: Key specifications (names or callables), applied in
            order and joined with ``|``.

    Returns:
        A single key function producing the joined result.

    Example:
        >>> from unittest.mock import Mock
        >>> request = Mock(user=Mock(is_authenticated=True, pk=42), META={}, headers={})
        >>> request.tenant_id = "acme"
        >>> combine("tenant", "user")(request)
        'tenant:acme|user:42'
    """
    resolved = [resolve_key_func(k) for k in key_funcs]

    def _combined(request: Any, view: Any = None) -> str:
        return "|".join(func(request, view) for func in resolved)

    return _combined

resolve_key_func

Resolve a key specification to a callable.

Parameters:

Name	Type	Description	Default
key	str | KeyFunc	Either a callable already matching the (request, view=None) -> str signature, or the name of a built-in key function: "ip", "user", "api_key", "tenant", or "view".	required

Returns:

Type	Description
KeyFunc	The resolved key function.

Raises:

Type	Description
InvalidKeyError	If key is a string that doesn't name a built-in key function.

Source code in src/drf_ratelimit_plus/keys.py

def resolve_key_func(key: str | KeyFunc) -> KeyFunc:
    """Resolve a key specification to a callable.

    Args:
        key: Either a callable already matching the ``(request, view=None)
            -> str`` signature, or the name of a built-in key function:
            ``"ip"``, ``"user"``, ``"api_key"``, ``"tenant"``, or
            ``"view"``.

    Returns:
        The resolved key function.

    Raises:
        drf_ratelimit_plus.exceptions.InvalidKeyError: If ``key`` is a
            string that doesn't name a built-in key function.
    """
    if callable(key):
        return key
    func = _BUILTIN_KEY_FUNCS.get(key)
    if func is None:
        raise InvalidKeyError(
            f"Unknown key function {key!r}. Built-in options: {sorted(_BUILTIN_KEY_FUNCS)}, "
            f"or pass a callable."
        )
    return func

Rates

Rate

A parsed rate limit: count requests per period seconds.

Attributes:

Name	Type	Description
count	int	The maximum number of requests (or weighted cost units) allowed per period.
period	float	The window length in seconds.

Source code in src/drf_ratelimit_plus/rates.py

@dataclass(frozen=True, slots=True)
class Rate:
    """A parsed rate limit: ``count`` requests per ``period`` seconds.

    Attributes:
        count: The maximum number of requests (or weighted cost units)
            allowed per ``period``.
        period: The window length in seconds.
    """

    count: int
    period: float

    @property
    def per_second(self) -> float:
        """The equivalent steady-state rate, in requests per second.

        Used by the token bucket algorithm as its refill rate.
        """
        return self.count / self.period

per_second property

The equivalent steady-state rate, in requests per second.

Used by the token bucket algorithm as its refill rate.

parse_rate

Parse a "<count>/<period>" string into a :class:Rate.

Parameters:

Name	Type	Description	Default
rate	str | Rate	Either an already-parsed :class:Rate (returned unchanged), or a string like "100/m", "1000/hour", "10/s". Accepted period spellings: s/sec/second, m/min/minute, h/hour, d/day.	required

Returns:

Type	Description
Rate	The parsed :class:Rate.

Raises:

Type	Description
InvalidRateError	If rate is a string that doesn't match the expected shape, or names an unrecognized period.

Example

parse_rate("100/m") Rate(count=100, period=60.0) parse_rate("10/s").per_second 10.0

Source code in src/drf_ratelimit_plus/rates.py

def parse_rate(rate: str | Rate) -> Rate:
    """Parse a ``"<count>/<period>"`` string into a :class:`Rate`.

    Args:
        rate: Either an already-parsed :class:`Rate` (returned unchanged),
            or a string like ``"100/m"``, ``"1000/hour"``, ``"10/s"``.
            Accepted period spellings: ``s``/``sec``/``second``,
            ``m``/``min``/``minute``, ``h``/``hour``, ``d``/``day``.

    Returns:
        The parsed :class:`Rate`.

    Raises:
        drf_ratelimit_plus.exceptions.InvalidRateError: If ``rate`` is a
            string that doesn't match the expected shape, or names an
            unrecognized period.

    Example:
        >>> parse_rate("100/m")
        Rate(count=100, period=60.0)
        >>> parse_rate("10/s").per_second
        10.0
    """
    if isinstance(rate, Rate):
        return rate
    match = _RATE_PATTERN.match(rate)
    if not match:
        raise InvalidRateError(
            f"Invalid rate string {rate!r}; expected '<count>/<period>', e.g. '100/m'."
        )
    count_str, period_name = match.groups()
    period_seconds = _PERIODS.get(period_name.lower())
    if period_seconds is None:
        raise InvalidRateError(
            f"Unrecognized period {period_name!r} in rate {rate!r}. "
            f"Valid periods: {sorted(set(_PERIODS))}."
        )
    count = int(count_str)
    if count <= 0:
        raise InvalidRateError(f"Rate count must be positive, got {count} in {rate!r}.")
    return Rate(count=count, period=float(period_seconds))

Tiers

default_tier_resolver

The built-in tier resolver: reads request.user.plan.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required

Returns:

Type	Description
str	The value of request.user.plan if the request has an
str	authenticated user with a plan attribute, otherwise the
str	DEFAULT_TIER setting.

Example

Attach a plan attribute (a string) to your user model, or override this entirely by passing your own tier_resolver callable — e.g. one that reads a related Subscription model.

Source code in src/drf_ratelimit_plus/tiers.py

def default_tier_resolver(request: Any) -> str:
    """The built-in tier resolver: reads ``request.user.plan``.

    Args:
        request: The incoming request.

    Returns:
        The value of ``request.user.plan`` if the request has an
        authenticated user with a ``plan`` attribute, otherwise the
        ``DEFAULT_TIER`` setting.

    Example:
        Attach a ``plan`` attribute (a string) to your user model, or
        override this entirely by passing your own ``tier_resolver``
        callable — e.g. one that reads a related ``Subscription`` model.
    """
    user = getattr(request, "user", None)
    plan = getattr(user, "plan", None) if user is not None else None
    return str(plan) if plan else str(get_setting("DEFAULT_TIER"))

resolve_tier_rate

Resolve which :class:~drf_ratelimit_plus.rates.Rate applies to a request.

Parameters:

Name	Type	Description	Default
request	Any	The incoming request.	required
tiers	Mapping[str, str | Rate]	A mapping of tier name to rate string (or already-parsed :class:~drf_ratelimit_plus.rates.Rate), e.g. {"free": "10/m", "pro": "100/m", "enterprise": "1000/m"}.	required
resolver	TierResolver	Determines which tier name applies to request. Defaults to :func:default_tier_resolver.	default_tier_resolver

Returns:

Type	Description
Rate	The resolved :class:~drf_ratelimit_plus.rates.Rate.

Raises:

Type	Description
UnknownTierError	If the resolved tier name has no entry in tiers and tiers also has no entry for the DEFAULT_TIER setting to fall back to.

Source code in src/drf_ratelimit_plus/tiers.py

def resolve_tier_rate(
    request: Any,
    tiers: Mapping[str, str | Rate],
    *,
    resolver: TierResolver = default_tier_resolver,
) -> Rate:
    """Resolve which :class:`~drf_ratelimit_plus.rates.Rate` applies to a request.

    Args:
        request: The incoming request.
        tiers: A mapping of tier name to rate string (or already-parsed
            :class:`~drf_ratelimit_plus.rates.Rate`), e.g.
            ``{"free": "10/m", "pro": "100/m", "enterprise": "1000/m"}``.
        resolver: Determines which tier name applies to ``request``.
            Defaults to :func:`default_tier_resolver`.

    Returns:
        The resolved :class:`~drf_ratelimit_plus.rates.Rate`.

    Raises:
        drf_ratelimit_plus.exceptions.UnknownTierError: If the resolved
            tier name has no entry in ``tiers`` and ``tiers`` also has no
            entry for the ``DEFAULT_TIER`` setting to fall back to.
    """
    tier_name = resolver(request)
    rate = tiers.get(tier_name)
    if rate is None:
        default_tier = get_setting("DEFAULT_TIER")
        rate = tiers.get(default_tier)
        if rate is None:
            raise UnknownTierError(
                f"No rate configured for tier {tier_name!r}, and no fallback "
                f"entry for the default tier {default_tier!r}. Configured "
                f"tiers: {sorted(tiers)}."
            )
    return parse_rate(rate)

Algorithms

fixed_window_check

Check a request against a fixed-window rate limit.

Counts requests in discrete, non-overlapping windows of rate.period seconds. Simple and cheap, but allows up to 2 * rate.count requests across a window boundary (e.g. a burst just before a window ends, followed immediately by another just after it starts). Prefer :func:sliding_window_check if that boundary burst is a concern.

Parameters:

Name	Type	Description	Default
client	Any	A Redis client (standalone or Cluster).	required
key	str	The fully-qualified Redis key for this rate limit identity.	required
rate	Rate	The configured limit.	required
cost	int	How many units this request consumes (for weighted costs).	1
now	float | None	Unused; accepted for calling-convention symmetry with the other algorithms.	None

Returns:

Name	Type	Description
The	LimitResult	class:~drf_ratelimit_plus.results.LimitResult.

Source code in src/drf_ratelimit_plus/algorithms.py

def fixed_window_check(
    client: Any, key: str, rate: Rate, *, cost: int = 1, now: float | None = None
) -> LimitResult:
    """Check a request against a fixed-window rate limit.

    Counts requests in discrete, non-overlapping windows of ``rate.period``
    seconds. Simple and cheap, but allows up to ``2 * rate.count`` requests
    across a window boundary (e.g. a burst just before a window ends,
    followed immediately by another just after it starts). Prefer
    :func:`sliding_window_check` if that boundary burst is a concern.

    Args:
        client: A Redis client (standalone or Cluster).
        key: The fully-qualified Redis key for this rate limit identity.
        rate: The configured limit.
        cost: How many units this request consumes (for weighted costs).
        now: Unused; accepted for calling-convention symmetry with the
            other algorithms.

    Returns:
        The :class:`~drf_ratelimit_plus.results.LimitResult`.
    """
    period_ms = int(rate.period * 1000)
    current, ttl_ms = client.eval(_FIXED_WINDOW_SCRIPT, 1, key, period_ms, cost)
    current = int(current)
    ttl_ms = int(ttl_ms)
    allowed = current <= rate.count
    remaining = max(0, rate.count - current)
    reset_seconds = math.ceil(ttl_ms / 1000)
    retry_after = reset_seconds if not allowed else None
    return LimitResult(
        allowed=allowed,
        limit=rate.count,
        remaining=remaining,
        reset_seconds=reset_seconds,
        retry_after=retry_after,
    )

sliding_window_check

Check a request against an approximate sliding-window rate limit.

Estimates the request count over the trailing rate.period seconds as a weighted blend of the current fixed window's count and the previous window's count, weighted by how far into the current window "now" is. This is the same approximation Cloudflare's public rate limiter documentation describes: memory-efficient (two counters per identity, not one entry per request) while avoiding fixed-window's boundary-burst problem.

Parameters:

Name	Type	Description	Default
client	Any	A Redis client (standalone or Cluster).	required
key	str	The fully-qualified Redis key for this rate limit identity. Both the current and previous window's actual keys are derived from this with a {hash-tag}-safe suffix, so they always land on the same Cluster slot.	required
rate	Rate	The configured limit.	required
cost	int	How many units this request consumes.	1
now	float | None	The current time, as seconds since the epoch. Defaults to :func:time.time; pass explicitly in tests for determinism.	None

Returns:

Name	Type	Description
The	LimitResult	class:~drf_ratelimit_plus.results.LimitResult. remaining
	LimitResult	and reset_seconds are based on the estimated (not exact)
	LimitResult	count, consistent with this algorithm's approximate nature.

Source code in src/drf_ratelimit_plus/algorithms.py

def sliding_window_check(
    client: Any, key: str, rate: Rate, *, cost: int = 1, now: float | None = None
) -> LimitResult:
    """Check a request against an approximate sliding-window rate limit.

    Estimates the request count over the trailing ``rate.period`` seconds
    as a weighted blend of the current fixed window's count and the
    previous window's count, weighted by how far into the current window
    "now" is. This is the same approximation Cloudflare's public rate
    limiter documentation describes: memory-efficient (two counters per
    identity, not one entry per request) while avoiding fixed-window's
    boundary-burst problem.

    Args:
        client: A Redis client (standalone or Cluster).
        key: The fully-qualified Redis key for this rate limit identity.
            Both the current and previous window's actual keys are
            derived from this with a ``{hash-tag}``-safe suffix, so they
            always land on the same Cluster slot.
        rate: The configured limit.
        cost: How many units this request consumes.
        now: The current time, as seconds since the epoch. Defaults to
            :func:`time.time`; pass explicitly in tests for determinism.

    Returns:
        The :class:`~drf_ratelimit_plus.results.LimitResult`. ``remaining``
        and ``reset_seconds`` are based on the estimated (not exact)
        count, consistent with this algorithm's approximate nature.
    """
    now_ms = int((now if now is not None else time.time()) * 1000)
    period_ms = int(rate.period * 1000)
    window_index = now_ms // period_ms
    elapsed_ms = now_ms - (window_index * period_ms)

    # Wrap the entire base key in a Redis Cluster hash tag ("{...}") so
    # only its content (not the differing window-index suffix appended
    # outside the braces) determines the hash slot - guaranteeing both
    # keys below always land on the same slot/node, with no cooperation
    # required from the caller.
    current_key = f"{{{key}}}:{window_index}"
    previous_key = f"{{{key}}}:{window_index - 1}"

    allowed_flag, estimated_total, _current, _previous = client.eval(
        _SLIDING_WINDOW_SCRIPT,
        2,
        current_key,
        previous_key,
        period_ms,
        cost,
        elapsed_ms,
        rate.count,
    )
    allowed = bool(int(allowed_flag))
    estimated_total = float(estimated_total)
    remaining = max(0, math.floor(rate.count - estimated_total))
    reset_seconds = math.ceil((period_ms - elapsed_ms) / 1000)
    retry_after = reset_seconds if not allowed else None
    return LimitResult(
        allowed=allowed,
        limit=rate.count,
        remaining=remaining,
        reset_seconds=reset_seconds,
        retry_after=retry_after,
    )

token_bucket_check

Check a request against a token-bucket rate limit.

The bucket refills continuously at rate.count / rate.period tokens per second, up to a capacity of burst (defaulting to rate.count if not given — i.e. no extra burst allowance beyond the steady-state rate). This is the algorithm to reach for when you want to allow a client to burst above their steady-state rate occasionally, as long as they've been under it recently (each unused token accumulates, up to the bucket's capacity).

Parameters:

Name	Type	Description	Default
client	Any	A Redis client (standalone or Cluster).	required
key	str	The fully-qualified Redis key for this rate limit identity — a single Redis Hash storing tokens and timestamp, so this is always a single-key (Cluster-safe) operation.	required
rate	Rate	The configured steady-state limit.	required
cost	int	How many tokens this request consumes.	1
burst	int | None	The bucket's capacity, i.e. the maximum number of tokens that can accumulate for later bursting. Defaults to rate.count.	None
now	float | None	The current time, as seconds since the epoch. Defaults to :func:time.time; pass explicitly in tests for determinism.	None

Returns:

Name	Type	Description
The	LimitResult	class:~drf_ratelimit_plus.results.LimitResult. limit is
	LimitResult	reported as the bucket's capacity (burst), since that's the
	LimitResult	maximum a single burst can consume.

Source code in src/drf_ratelimit_plus/algorithms.py

def token_bucket_check(
    client: Any,
    key: str,
    rate: Rate,
    *,
    cost: int = 1,
    burst: int | None = None,
    now: float | None = None,
) -> LimitResult:
    """Check a request against a token-bucket rate limit.

    The bucket refills continuously at ``rate.count / rate.period`` tokens
    per second, up to a capacity of ``burst`` (defaulting to ``rate.count``
    if not given — i.e. no extra burst allowance beyond the steady-state
    rate). This is the algorithm to reach for when you want to allow a
    client to burst above their steady-state rate occasionally, as long as
    they've been under it recently (each unused token accumulates, up to
    the bucket's capacity).

    Args:
        client: A Redis client (standalone or Cluster).
        key: The fully-qualified Redis key for this rate limit identity —
            a single Redis Hash storing ``tokens`` and ``timestamp``, so
            this is always a single-key (Cluster-safe) operation.
        rate: The configured steady-state limit.
        cost: How many tokens this request consumes.
        burst: The bucket's capacity, i.e. the maximum number of tokens
            that can accumulate for later bursting. Defaults to
            ``rate.count``.
        now: The current time, as seconds since the epoch. Defaults to
            :func:`time.time`; pass explicitly in tests for determinism.

    Returns:
        The :class:`~drf_ratelimit_plus.results.LimitResult`. ``limit`` is
        reported as the bucket's capacity (``burst``), since that's the
        maximum a single burst can consume.
    """
    capacity = burst if burst is not None else rate.count
    refill_per_ms = rate.per_second / 1000.0
    now_ms = int((now if now is not None else time.time()) * 1000)
    # Keep the bucket alive long enough to refill from empty to capacity,
    # plus a safety margin, so a quiet identity doesn't linger forever but
    # also doesn't expire mid-accumulation.
    ttl_ms = int((capacity / rate.per_second) * 1000) + int(rate.period * 1000)

    allowed_flag, tokens_str = client.eval(
        _TOKEN_BUCKET_SCRIPT, 1, key, capacity, refill_per_ms, cost, now_ms, ttl_ms
    )
    allowed = bool(int(allowed_flag))
    tokens = float(tokens_str)
    remaining = max(0, math.floor(tokens))
    if allowed:
        retry_after = None
    else:
        deficit = cost - tokens
        retry_after = math.ceil(deficit / refill_per_ms / 1000) if refill_per_ms > 0 else None
    reset_seconds = math.ceil((capacity - tokens) / rate.per_second) if rate.per_second > 0 else 0
    return LimitResult(
        allowed=allowed,
        limit=capacity,
        remaining=remaining,
        reset_seconds=max(0, reset_seconds),
        retry_after=retry_after,
    )

Results

LimitResult

The outcome of checking one request against a rate limit.

Attributes:

Name	Type	Description
allowed	bool	Whether the request is within its limit.
limit	int	The configured maximum (requests, or weighted cost units, per window/bucket capacity).
remaining	int	How much of the limit is left after this check (never negative).
reset_seconds	int	Seconds until the limit fully resets (the window closes, or the bucket refills to capacity). Always present, regardless of allowed.
retry_after	int | None	Seconds the caller should wait before retrying. None when allowed is True.

Source code in src/drf_ratelimit_plus/results.py

@dataclass(frozen=True, slots=True)
class LimitResult:
    """The outcome of checking one request against a rate limit.

    Attributes:
        allowed: Whether the request is within its limit.
        limit: The configured maximum (requests, or weighted cost units,
            per window/bucket capacity).
        remaining: How much of the limit is left after this check
            (never negative).
        reset_seconds: Seconds until the limit fully resets (the window
            closes, or the bucket refills to capacity). Always present,
            regardless of ``allowed``.
        retry_after: Seconds the caller should wait before retrying.
            ``None`` when ``allowed`` is ``True``.
    """

    allowed: bool
    limit: int
    remaining: int
    reset_seconds: int
    retry_after: int | None

Client

get_redis_client

Return the configured Redis client, constructing it on first use.

Resolution order:

1. If the REDIS_CLIENT setting is set, import it — it may be a dotted path to a client instance, or to a no-argument callable returning one.
2. Otherwise, build a new :class:redis.Redis from the REDIS_URL setting.

The result is cached for the lifetime of the process (invalidated automatically if RATELIMIT_PLUS changes, e.g. via @override_settings in tests).

Returns:

Type	Description
Any	A Redis client (standalone or Cluster) implementing at least
Any	.eval().

Source code in src/drf_ratelimit_plus/client.py

def get_redis_client() -> Any:
    """Return the configured Redis client, constructing it on first use.

    Resolution order:

    1. If the ``REDIS_CLIENT`` setting is set, import it — it may be a
       dotted path to a client instance, or to a no-argument callable
       returning one.
    2. Otherwise, build a new :class:`redis.Redis` from the ``REDIS_URL``
       setting.

    The result is cached for the lifetime of the process (invalidated
    automatically if ``RATELIMIT_PLUS`` changes, e.g. via
    ``@override_settings`` in tests).

    Returns:
        A Redis client (standalone or Cluster) implementing at least
        ``.eval()``.
    """
    global _client
    if _client is None:
        dotted_path = get_setting("REDIS_CLIENT")
        if dotted_path:
            candidate = import_string(dotted_path)
            _client = candidate() if callable(candidate) else candidate
        else:
            _client = redis.Redis.from_url(get_setting("REDIS_URL"))
    return _client

set_redis_client

Override the cached client directly (primarily for tests).

Parameters:

Name	Type	Description	Default
client	Any | None	The client to use for subsequent calls to :func:get_redis_client, or None to clear the cache and force it to be rebuilt from settings on next use.	required

Source code in src/drf_ratelimit_plus/client.py

def set_redis_client(client: Any | None) -> None:
    """Override the cached client directly (primarily for tests).

    Args:
        client: The client to use for subsequent calls to
            :func:`get_redis_client`, or ``None`` to clear the cache and
            force it to be rebuilt from settings on next use.
    """
    global _client
    _client = client

Exceptions

RateLimitPlusError

Bases: Exception

Base class for all configuration errors raised by this package.

Source code in src/drf_ratelimit_plus/exceptions.py

class RateLimitPlusError(Exception):
    """Base class for all configuration errors raised by this package."""

InvalidRateError

Bases: RateLimitPlusError, ValueError

Raised when a rate string cannot be parsed.

See :func:drf_ratelimit_plus.rates.parse_rate.

Source code in src/drf_ratelimit_plus/exceptions.py

class InvalidRateError(RateLimitPlusError, ValueError):
    """Raised when a rate string cannot be parsed.

    See :func:`drf_ratelimit_plus.rates.parse_rate`.
    """

InvalidKeyError

Bases: RateLimitPlusError, ValueError

Raised when a key argument names an unknown built-in key function.

See :func:drf_ratelimit_plus.keys.resolve_key_func.

Source code in src/drf_ratelimit_plus/exceptions.py

class InvalidKeyError(RateLimitPlusError, ValueError):
    """Raised when a ``key`` argument names an unknown built-in key function.

    See :func:`drf_ratelimit_plus.keys.resolve_key_func`.
    """

InvalidAlgorithmError

Bases: RateLimitPlusError, ValueError

Raised when an algorithm argument names an unknown algorithm.

See :func:drf_ratelimit_plus.throttles.rate_limit.

Source code in src/drf_ratelimit_plus/exceptions.py

class InvalidAlgorithmError(RateLimitPlusError, ValueError):
    """Raised when an ``algorithm`` argument names an unknown algorithm.

    See :func:`drf_ratelimit_plus.throttles.rate_limit`.
    """

UnknownTierError

Bases: RateLimitPlusError, ValueError

Raised when a resolved plan-tier name has no matching entry.

See :mod:drf_ratelimit_plus.tiers.

Source code in src/drf_ratelimit_plus/exceptions.py

class UnknownTierError(RateLimitPlusError, ValueError):
    """Raised when a resolved plan-tier name has no matching entry.

    See :mod:`drf_ratelimit_plus.tiers`.
    """

Settings

See Settings for the full list of recognized RATELIMIT_PLUS keys.