Throttling

django-modern-rest ships its own throttling (also known as “rate limiting”) mechanism. Here’s how everything works.

Important

If you have an option not to use ratelimiting in Django, but to use it on the HTTP Proxy side, you should prefer the proxy. It is significantly faster and more secure.

Defining throttling

We have two classes to define throttling:

• dmr.throttling.SyncThrottle for sync endpoints

• dmr.throttling.AsyncThrottle for async endpoints

We can define throttling on three different levels:

per endpointper controllerper settings

views.py

Show imports... 5 lines hidden

[source]

from dmr import Controller, modify
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import Rate, SyncThrottle


class SyncController(Controller[PydanticSerializer]):
    @modify(throttling=[SyncThrottle(1, Rate.minute)])
    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
"inside"

$ curl http://127.0.0.1:8000/api/sync/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Fri, 05 Jun 2026 12:23:01 GMT
server: uvicorn
X-RateLimit-Limit: 1
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 60
Retry-After: 60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 59
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Too many requests","type":"ratelimit"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "Retry-After": {
                "description": "Indicates how long the user agent should wait before making a follow-up request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

Providing several throttling instances means that all of them must succeed. When multiple throttling rules are defined on different levels, their rules are joined.

For example:

views.py

Show imports... 5 lines hidden

[source]

from dmr import Controller, modify
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import Rate, SyncThrottle


class SyncController(Controller[PydanticSerializer]):
    throttling = (SyncThrottle(5, Rate.hour),)

    @modify(throttling=[SyncThrottle(1, Rate.minute)])
    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
"inside"

$ curl http://127.0.0.1:8000/api/sync/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Fri, 05 Jun 2026 12:23:02 GMT
server: uvicorn
X-RateLimit-Limit: 1
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 59
Retry-After: 59
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 59
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Too many requests","type":"ratelimit"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "Retry-After": {
                "description": "Indicates how long the user agent should wait before making a follow-up request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

Will guard GET method with 2 throttling checks:

1. Not more <= than 1 request per minute

2. And not more <= than 5 requests per hour

Customizing throttling

Rates

Rate is passed as the second required parameter to throttle classes. However, all values that you pass are just numbers of seconds. So, you can fully customize throttling timings by passing any amount of seconds that you wish:

settings.py

>>> from dmr.settings import Settings, DMR_SETTINGS
>>> from dmr.throttling import SyncThrottle

>>> DMR_SETTINGS = {
...     Settings.throttling: [
...          SyncThrottle(
...              max_requests=5,
...              duration_in_seconds=10,
...          ),
...     ],
... }

This will set a throttling rule: no more than 5 requests in 10 seconds.

Backends

Backends are used to define where we store throttling data.

By default we use:

• dmr.throttling.backends.SyncDjangoCache for sync endpoints

• dmr.throttling.backends.AsyncDjangoCache for async endpoints

All backends that we support can be further customized.

DjangoCacheRedis

By default we store all the data in the 'default' Django cache. You can customize which Django cache name is used. For example:

views.py

Show imports... 6 lines hidden

[source]

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.backends import SyncDjangoCache


class SyncController(Controller[PydanticSerializer]):
    throttling = (
        SyncThrottle(
            max_requests=1,
            duration_in_seconds=Rate.minute,
            backend=SyncDjangoCache(cache_name='throttling'),
        ),
    )

    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
"inside"

$ curl http://127.0.0.1:8000/api/sync/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Fri, 05 Jun 2026 12:23:03 GMT
server: uvicorn
X-RateLimit-Limit: 1
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 60
Retry-After: 60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 59
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Too many requests","type":"ratelimit"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "Retry-After": {
                "description": "Indicates how long the user agent should wait before making a follow-up request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

You can also write your own backends, for example, to store throttling information in memory, filesystem, or somewhere else. To do so, you would need to subclass dmr.throttling.backends.BaseThrottleSyncBackend or dmr.throttling.backends.BaseThrottleAsyncBackend and override 2 methods.

Full list of backends that we ship in django-modern-rest:

• SyncDjangoCache and AsyncDjangoCache, default

• SyncRedis and AsyncRedis

Warning

When using SyncDjangoCache or AsyncDjangoCache the final behavior will depend on the cache that you use.

Some Django cache backends like django.core.cache.backends.locmem.LocMemCache store cache in memory per-process. So, any multiprocess environments with N processes will allow to use N * max_request requests. Using such cache backends is not safe.

Some like django.core.cache.backends.dummy.DummyCache do nothing at all.

Choosing a backend

Backend	Atomicity	Overhead	Supported algorithms	Best suited for
DjangoCache	Per-process: multiprocess deployments may face problems	Very low (depends on the cache type)	All	Non-critical IP based checks with not-strict windows and limits
Redis	Full	Low	All builtin ones, but requires lua scripting support	Strict distributed limits

Unsafe backend warning

By default, django-modern-rest emits a UnsafeCacheBackendWarning warning when detecting an unsafe cache backend for throttling.

You can configure this check on three levels as usual:

1. Per endpoint: pass throttling_allow_unsafe_cache parameter

2. Per controller: by setting throttling_allow_unsafe_cache attribute

3. In settings, see throttling_allow_unsafe_cache

When throttling_allow_unsafe_cache is set to False, we raise a dmr.exceptions.EndpointMetadataError exception instead of a warning. This setting will ensure the maximum safety.

To suppress this check completely and run throttling at your own risk, set throttling_allow_unsafe_cache to None.

Algorithms

Algorithms are used to define the logic of how requests are counted.

By default we use dmr.throttling.algorithms.SimpleRate as the algorithm. It defines a fixed window with a fixed amount of requests possible. When window is expired, it resets the count of requests.

Here’s how you can customize the algorithm for a throttling:

views.py

Show imports... 6 lines hidden

[source]

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.algorithms import LeakyBucket


class SyncController(Controller[PydanticSerializer]):
    throttling = (
        SyncThrottle(
            max_requests=1,
            duration_in_seconds=Rate.minute,
            algorithm=LeakyBucket(),
        ),
    )

    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
"inside"

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "Retry-After": {
                "description": "Indicates how long the user agent should wait before making a follow-up request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

You can also write your own algorithms. To do so, you would need to subclass dmr.throttling.algorithms.BaseThrottleAlgorithm and override 2 required methods.

Full list of algorithms that we ship in django-modern-rest:

• SimpleRate, default

• LeakyBucket where requests fill the bucket; tokens leak at a steady rate. Unlike SimpleRate, drains continuously providing smoother rate-limiting without allowing bursts at window boundaries.

Warning

SimpleRate uses a fixed window that resets when the window expires. This allows a boundary burst pattern: a client can send N requests right before the window resets and N more right after, effectively firing 2N requests in a short interval while remaining within the configured per-window limit.

For abuse-sensitive endpoints (login, OTP, password reset) prefer LeakyBucket, which drains continuously and eliminates this burst window.

Choosing an algorithm

Algorithm	Window type	Overhead	Boundary burst risk	Best suited for
SimpleRate	Fixed — resets after duration	Very low	Yes — up to 2N requests in a short burst	General-purpose, internal, and admin endpoints
LeakyBucket	Continuous drain	Low	No — traffic is smoothed regardless of timing	Auth endpoints (login, OTP, password reset), public APIs

For auth and abuse-sensitive endpoints, use LeakyBucket:

views.py

Show imports... 6 lines hidden

[source]

from dmr import Controller
from dmr.plugins.pydantic import PydanticFastSerializer
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.algorithms import LeakyBucket


class LoginController(Controller[PydanticFastSerializer]):
    throttling = (
        SyncThrottle(
            max_requests=5,
            duration_in_seconds=Rate.minute,
            algorithm=LeakyBucket(),
        ),
    )

    def post(self) -> str:
        return 'logged in'

Cache keys

Cache keys is what defines how requests are identified.

By default we use dmr.throttling.cache_keys.RemoteAddr() cache key, which identifies requests by IP taken from REMOTE_ADDR value from request.META.

Warning

If you are using reverse proxies, make sure to correctly configure how they pass request headers, to REMOTE_ADDR would be correct.

You can write your own cache keys, they are subclasses of BaseThrottleCacheKey and must return a string or None.

If cache key returns None, it means that this request will be skipped from this exact throttling check. However, other keys may still be applied.

It is useful to skip some requests from throttling checks, for example, from paid or stuff users.

Full list of cache keys that we ship in django-modern-rest:

• RemoteAddr, default

• UserPk, based on request.user, by default we use request.user.pk if it exists. You can pass exclude_stuff argument as False to also limit is_stuff users, or you can pass exclude_superuser argument as False to also limit super users

• JwtToken, based on request.__dmr_jwt__. Uses jti claim when present and falls back to sub claim. Raw value is hashed before being used as a cache key. Returns None when request.__dmr_jwt__ is not set.

When throttling is executed

Throttling is executed in two stages: before auth and after auth. Why? Because we need to:

1. Protect auth from abusive requests and brute forcing

2. Make sure we can base throttling rules on the auth info

The same can be said about content negotiation, it also must be protected by throttling. Otherwise, people can abuse content negotiation without any request limits.

        ---
config:
  theme: forest

---
graph
    Start[New request] --> BeforeThrottle[Throttling based on IP or 429];
    BeforeThrottle --> RendererNegotiation[Renderer is negotiated or 406];
    RendererNegotiation --> Auth[Auth or 401];
    Auth --> AfterThrottle[Throttling based on auth or 429];
    

Throttling execution

All cache keys know when to execute by default, however you can customize this. For example, you can run some IP based throttling checks after the auth itself:

views.py

Show imports... 7 lines hidden

[source]

from dmr import Controller, modify
from dmr.plugins.pydantic import PydanticSerializer
from dmr.security.django_session import DjangoSessionSyncAuth
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.cache_keys import RemoteAddr


class SyncController(Controller[PydanticSerializer]):
    @modify(
        auth=[DjangoSessionSyncAuth()],
        throttling=[
            SyncThrottle(
                1,
                Rate.minute,
                cache_key=RemoteAddr(runs_before_auth=False),
            ),
        ],
    )
    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
{"detail":[{"msg":"Not authenticated","type":"security"}]}

$ curl http://127.0.0.1:8000/api/sync/ -X GET
{"detail":[{"msg":"Not authenticated","type":"security"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {
      "csrf": {
        "description": "CSRF protection",
        "in": "cookie",
        "name": "csrftoken",
        "type": "apiKey"
      },
      "django_session": {
        "description": "Reusing standard Django auth flow for API",
        "in": "cookie",
        "name": "sessionid",
        "type": "apiKey"
      }
    }
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "401": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when auth was not successful"
          },
          "403": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when CSRF check failed"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "Retry-After": {
                "description": "Indicates how long the user agent should wait before making a follow-up request",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        },
        "security": [
          {
            "csrf": [],
            "django_session": []
          }
        ]
      }
    }
  }
}

Warning

It is strongly not recommended to have auth without any throttling before it.

Auth must be protected from brute force and denial of service attacks! For example, one can also use django-axes for this.

wemake-django-template has this configured properly.

Note that it won’t make any sense to run auth-based throttling before auth. So, customize it with care.

Headers

By default on 429 Too Many Requests error we return four headers:

• X-RateLimit-Limit - The maximum number of requests permitted in the current time window

• X-RateLimit-Remaining - The number of requests remaining in the current time window

• X-RateLimit-Reset - The number of seconds until the current rate limit window resets

• Retry-After - The number of seconds until the current rate limit window resets, see RFC-6585 and RFC-7231

Note

Headers with X- prefix means that they are custom ones, there’s no spec behind them. However, this convention is the most popular one as of right now.

OpenAPI support is built in for this feature. All headers classes will provide the proper HeaderSpec for the 429 response.

You might want to customize the returned headers. To do so, you can pass response_headers argument to throttling classes with header classes that you actually want to support.

For example, you can disable Retry-After header with:

views.py

Show imports... 6 lines hidden

[source]

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import AsyncThrottle, Rate
from dmr.throttling.headers import XRateLimit


class AsyncController(Controller[PydanticSerializer]):
    throttling = (
        AsyncThrottle(1, Rate.minute, response_headers=[XRateLimit()]),
    )

    async def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/async/ -X GET
"inside"

$ curl http://127.0.0.1:8000/api/async/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Fri, 05 Jun 2026 12:23:06 GMT
server: uvicorn
X-RateLimit-Limit: 1
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 59
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Too many requests","type":"ratelimit"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/asynccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getAsynccontrollerApiAsynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "X-RateLimit-Limit": {
                "description": "The maximum number of requests permitted in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Remaining": {
                "description": "The number of requests remaining in the current time window",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "X-RateLimit-Reset": {
                "description": "The number of seconds until the current rate limit window resets",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

Or if you want to support the latest draft about RateLimit and RateLimit-Policy headers, you can use:

views.py

Show imports... 6 lines hidden

[source]

from dmr import Controller
from dmr.plugins.pydantic import PydanticSerializer
from dmr.throttling import Rate, SyncThrottle
from dmr.throttling.headers import RateLimitIETFDraft


class SyncController(Controller[PydanticSerializer]):
    throttling = (
        SyncThrottle(1, Rate.minute, response_headers=[RateLimitIETFDraft()]),
    )

    def get(self) -> str:
        return 'inside'

Run result

$ curl http://127.0.0.1:8000/api/sync/ -X GET
"inside"

$ curl http://127.0.0.1:8000/api/sync/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Fri, 05 Jun 2026 12:23:07 GMT
server: uvicorn
RateLimit-Policy: 1;w=60;name="RemoteAddr"
RateLimit: "RemoteAddr";r=0;t=59
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 59
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Too many requests","type":"ratelimit"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK"
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

You can also combine these headers with each other in any combinations. You can write your own classes with custom headers support. To do so, subclass dmr.throttling.headers.BaseResponseHeadersProvider.

You can completely disable any extra response headers by passing an empty list.

Full list of header providers that we ship in django-modern-rest:

• XRateLimit, default

• RetryAfter, default

• RateLimitIETFDraft

Security

Key considerations:

• Be sure to correctly setup your HTTP Proxy server to send correct IP headers

• Be especially careful with X-Forwarded-For header, because it can contain several layers of proxies

• Never rate limit on user supplied data such as User-Agent, because this data can easily be changed

• Denial of service: be careful not to limit other users when limiting just one

• Do not store sensitive or personal users’ data in your cache keys, because it is stored with no protection / encryption

See also

https://django-ratelimit.readthedocs.io/en/latest/security.html

Throttling reports

If you need to attach any throttling headers to successful responses, you can do it as well.

For this we offer two APIs:

• dmr.throttling.ThrottlingReport.report() for sync APIs

• dmr.throttling.ThrottlingReport.areport() for async APIs

All our regular rules apply:

• All new headers must be added to the corresponding ResponseSpec definitions

• When settings headers, you would need to use validate()

views.py

Show imports... 11 lines hidden

[source]

from http import HTTPStatus
from typing import Final

from django.http import HttpResponse

from dmr import Controller, ResponseSpec, validate
from dmr.plugins.pydantic import PydanticFastSerializer
from dmr.throttling import AsyncThrottle, Rate, ThrottlingReport
from dmr.throttling.cache_keys import RemoteAddr
from dmr.throttling.headers import RateLimitIETFDraft

_draft_headers: Final = RateLimitIETFDraft()


class AsyncController(Controller[PydanticFastSerializer]):
    @validate(
        ResponseSpec(
            str,
            status_code=HTTPStatus.OK,
            headers=_draft_headers.provide_headers_specs(),
        ),
        throttling=[
            AsyncThrottle(
                1,
                Rate.second,
                response_headers=[_draft_headers],
                cache_key=RemoteAddr(name='per-second'),
            ),
            AsyncThrottle(
                5,
                Rate.minute,
                response_headers=[_draft_headers],
                cache_key=RemoteAddr(name='per-minute'),
            ),
        ],
    )
    async def get(self) -> HttpResponse:
        return self.to_response(
            'inside',
            headers=await ThrottlingReport(self).areport(),
        )

Run result

$ curl http://127.0.0.1:8000/api/async/ -D - -X GET
HTTP/1.1 200 OK
date: Fri, 05 Jun 2026 12:23:07 GMT
server: uvicorn
RateLimit-Policy: 1;w=1;name="per-second", 5;w=60;name="per-minute"
RateLimit: "per-second";r=0;t=1, "per-minute";r=4;t=60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 8
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

"inside"

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/asynccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getAsynccontrollerApiAsynccontroller",
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied"
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema"
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

Use headers argument to to_response() to add needed headers.

Warning

ThrottlingReport will make N cache requests when building header reports (where N is the number of throttle instances used for this endpoint).

It might be slow, depending on the number of throttles and your cache.

It might also fail, we don’t handle any errors in the reports building process.

Use this feature only when there’s a serious need for it.

You can also provide the same headers not just for successful responses, but for any errors that consume ratelimit quota as well.

To so, you would need to customize headers spec of your error model:

views.py

Show imports... 17 lines hidden

[source]

from collections.abc import Mapping
from http import HTTPStatus
from typing import Annotated, Any, Final

import pydantic
from django.http import HttpResponse
from typing_extensions import override

from dmr import Controller, NewCookie, Query, ResponseSpec, validate
from dmr.errors import ErrorModel
from dmr.metadata import ResponseSpecMetadata
from dmr.plugins.pydantic import PydanticFastSerializer
from dmr.renderers import Renderer
from dmr.throttling import Rate, SyncThrottle, ThrottlingReport
from dmr.throttling.cache_keys import RemoteAddr
from dmr.throttling.headers import RateLimitIETFDraft

_draft_headers: Final = RateLimitIETFDraft()


class _QueryModel(pydantic.BaseModel):
    number: int


class SyncController(Controller[PydanticFastSerializer]):
    error_model = Annotated[
        ErrorModel,
        ResponseSpecMetadata(headers=_draft_headers.provide_headers_specs()),
    ]

    @validate(
        ResponseSpec(
            str,
            status_code=HTTPStatus.OK,
            headers=_draft_headers.provide_headers_specs(),
        ),
        throttling=[
            SyncThrottle(
                2,
                Rate.second,
                response_headers=[_draft_headers],
                cache_key=RemoteAddr(name='per-second'),
            ),
            SyncThrottle(
                5,
                Rate.minute,
                response_headers=[_draft_headers],
                cache_key=RemoteAddr(name='per-minute'),
            ),
        ],
    )
    def get(self, parsed_query: Query[_QueryModel]) -> HttpResponse:
        return self.to_response('inside')

    @override
    def to_response(
        self,
        raw_data: Any,
        *,
        status_code: HTTPStatus | None = None,
        headers: Mapping[str, str] | None = None,
        cookies: Mapping[str, NewCookie] | None = None,
        renderer: Renderer | None = None,
    ) -> HttpResponse:
        response_headers = ThrottlingReport(self).report()
        response_headers.update(headers or {})
        return super().to_response(
            raw_data,
            status_code=status_code,
            headers=response_headers,
            cookies=cookies,
            renderer=renderer,
        )

Run result

$ curl 'http://127.0.0.1:8000/api/sync/?number=1' -D - -X GET
HTTP/1.1 200 OK
date: Fri, 05 Jun 2026 12:23:08 GMT
server: uvicorn
RateLimit-Policy: 2;w=1;name="per-second", 5;w=60;name="per-minute"
RateLimit: "per-second";r=1;t=1, "per-minute";r=4;t=60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 8
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

"inside"

$ curl 'http://127.0.0.1:8000/api/sync/?number=a' -D - -X GET
HTTP/1.1 400 Bad Request
date: Fri, 05 Jun 2026 12:23:08 GMT
server: uvicorn
RateLimit-Policy: 2;w=1;name="per-second", 5;w=60;name="per-minute"
RateLimit: "per-second";r=0;t=1, "per-minute";r=3;t=60
Content-Type: application/json
X-Frame-Options: DENY
Vary: Accept-Language
Content-Language: en
Content-Length: 145
X-Content-Type-Options: nosniff
Referrer-Policy: same-origin
Cross-Origin-Opener-Policy: same-origin

{"detail":[{"msg":"Input should be a valid integer, unable to parse string as an integer","loc":["parsed_query","number"],"type":"value_error"}]}

OpenAPI Schema

Preview openapi.json

{
  "components": {
    "schemas": {
      "ErrorDetail": {
        "description": "Base schema for error details description.",
        "properties": {
          "loc": {
            "items": {
              "anyOf": [
                {
                  "type": "integer"
                },
                {
                  "type": "string"
                }
              ]
            },
            "title": "Loc",
            "type": "array"
          },
          "msg": {
            "title": "Msg",
            "type": "string"
          },
          "type": {
            "title": "Type",
            "type": "string"
          }
        },
        "required": [
          "msg"
        ],
        "title": "ErrorDetail",
        "type": "object"
      },
      "ErrorModel": {
        "description": "Default error response schema.\n\nCan be customized.\nSee :ref:`customizing-error-messages` for more details.",
        "properties": {
          "detail": {
            "items": {
              "$ref": "#/components/schemas/ErrorDetail"
            },
            "title": "Detail",
            "type": "array"
          }
        },
        "required": [
          "detail"
        ],
        "title": "ErrorModel",
        "type": "object"
      }
    },
    "securitySchemes": {}
  },
  "info": {
    "title": "Django Modern Rest",
    "version": "0.1.0"
  },
  "openapi": "3.2.0",
  "paths": {
    "/api/synccontroller/": {
      "get": {
        "deprecated": false,
        "operationId": "getSynccontrollerApiSynccontroller",
        "parameters": [
          {
            "deprecated": false,
            "in": "query",
            "name": "number",
            "required": true,
            "schema": {
              "title": "Number",
              "type": "integer"
            }
          }
        ],
        "responses": {
          "200": {
            "content": {
              "application/json": {
                "schema": {
                  "type": "string"
                }
              }
            },
            "description": "OK",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "400": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when request components cannot be parsed",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "406": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when provided `Accept` header cannot be satisfied",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "422": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when returned response does not match the response schema",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          },
          "429": {
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ErrorModel"
                }
              }
            },
            "description": "Raised when throttling rate was hit",
            "headers": {
              "RateLimit": {
                "description": "Current rate limiting state",
                "required": true,
                "schema": {
                  "type": "string"
                }
              },
              "RateLimit-Policy": {
                "description": "Description of all rate limiting policies for this endpoint",
                "required": true,
                "schema": {
                  "type": "string"
                }
              }
            }
          }
        }
      }
    }
  }
}

Method to_response is used for both successful and error responses. This way both your successful responses and error responses will have the needed ratelimiting headers.

API Reference

Base

class dmr.throttling.SyncThrottle(max_requests: int, duration_in_seconds: Rate | int, *, cache_key: BaseThrottleCacheKey | None = None, backend: _BackendT | None = None, algorithm: BaseThrottleAlgorithm | None = None, response_headers: Iterable[BaseResponseHeadersProvider] | None = None)

Sync throttle type for sync endpoints.

Added in version 0.7.0.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer], lock: AbstractContextManager[Any, Any]) → None

Put your throttle business logic here.

Return None if throttle check passed. Raise dmr.exceptions.TooManyRequestsError if throttle check failed. Raise dmr.response.APIError if you want to change the return code, for example, when some data is missing or has wrong format.

collect_response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], remaining: int, reset: int, *, report_all: bool = False) → dict[str, str]

Collects response headers for all response_headers classes.

full_cache_key(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Get the full cache key value.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Provides responses that can happen when throttle triggers.

report_usage(endpoint: Endpoint, controller: Controller[BaseSerializer]) → dict[str, str]

Report throttle usage stats.

class dmr.throttling.AsyncThrottle(max_requests: int, duration_in_seconds: Rate | int, *, cache_key: BaseThrottleCacheKey | None = None, backend: _BackendT | None = None, algorithm: BaseThrottleAlgorithm | None = None, response_headers: Iterable[BaseResponseHeadersProvider] | None = None)

Async throttle type for async endpoints.

Added in version 0.7.0.

async __call__(endpoint: Endpoint, controller: Controller[BaseSerializer], lock: AbstractAsyncContextManager[Any, Any]) → None

Put your throttle business logic here.

Return None if throttle check passed. Raise dmr.exceptions.TooManyRequestsError if throttle check failed. Raise dmr.response.APIError if you want to change the return code, for example, when some data is missing or has wrong format.

collect_response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], remaining: int, reset: int, *, report_all: bool = False) → dict[str, str]

Collects response headers for all response_headers classes.

full_cache_key(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Get the full cache key value.

provide_response_specs(metadata: EndpointMetadata, controller_cls: type[Controller[BaseSerializer]], existing_responses: Mapping[HTTPStatus, ResponseSpec]) → list[ResponseSpec]

Provides responses that can happen when throttle triggers.

async report_usage(endpoint: Endpoint, controller: Controller[BaseSerializer]) → dict[str, str]

Async report throttle usage stats.

final class dmr.throttling.Rate(*values)

Throttling rates in seconds.

second

1 second.

minute

60 seconds.

hour

60 * 60 seconds.

day

24 * 60 * 60 seconds.

class dmr.throttling.ThrottlingReport(controller: Controller[BaseSerializer])

Get throttling data to be reported in response headers.

Unlike dmr.exceptions.TooManyRequestsError, which reports the first stat for throttle that is failing, it reports all stats for all throttles.

It will make N (which is the number of throttles) requests to cache.

Warning

This class does not handle exceptions at all. Because we don’t know how to report headers for failing cache connections. If you want to handle errors, catch them explicitly.

Added in version 0.7.0.

async areport() → dict[str, str]

Report throttling all headers for a async controller and endpoint.

Unlike sync version it makes parallel requests to cache by utilizing asyncio.gather() function inside.

report() → dict[str, str]

Report throttling all headers for a sync controller and endpoint.

Note that it will make N consecutive requests to cache. It might be rather long. Use the sync version with care.

Backends

class dmr.throttling.backends.CachedRateLimit

Bases: TypedDict

Representation of a cached object’s metadata.

class dmr.throttling.backends.BaseThrottleSyncBackend

Base class for all throttling backends.

It must provide sync and async API for sync and async throttling classes.

abstractmethod get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str) → CachedRateLimit | None

Sync get the state with no increments.

abstractmethod incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Sync increment cached rate limit state.

Can be atomic, can be non atomic. Atomicity needs to be documented.

class dmr.throttling.backends.BaseThrottleAsyncBackend

Base class for all throttling backends.

It must provide sync and async API for sync and async throttling classes.

abstractmethod async get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str) → CachedRateLimit | None

Sync get the state with no increments.

abstractmethod async incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Async increment cached rate limit state.

class dmr.throttling.backends.SyncDjangoCache(cache_name: str = 'default')

Uses Django sync cache framework for storing the rate limiting state.

See also

https://docs.djangoproject.com/en/stable/topics/cache/

get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str) → CachedRateLimit | None

Sync get the cached rate limit state.

incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Sync increment cached rate limit state.

Can be atomic, can be non atomic. Atomicity needs to be documented.

class dmr.throttling.backends.AsyncDjangoCache(cache_name: str = 'default')

Uses Django async cache framework for storing the rate limiting state.

See also

https://docs.djangoproject.com/en/stable/topics/cache/

async get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str) → CachedRateLimit | None

Async get the cached rate limit state.

async incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Async increment cached rate limit state.

final class dmr.throttling.backends.django_cache.UnsafeCacheBackendWarning

Warning emitted when an unsafe cache backend is used for throttling.

class dmr.throttling.backends.redis.SyncRedis(client: redis.Redis[Any])

Uses sync Redis client for multiproccess safe rate-limiting.

This backend requires redis client library to be installed. We don’t ship it with the django-modern-rest.

You would have to install it separately:

pip install redis

See also

https://redis.readthedocs.io

get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str) → CachedRateLimit | None

Sync get the cached rate limit state.

incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Sync increment cached rate limit state.

Can be atomic, can be non atomic. Atomicity needs to be documented.

initialize_algorithm(algorithm: BaseThrottleAlgorithm) → None

Initialize and prepare backend for the algorithm.

needs_transaction_script: ClassVar[str] = 'lua'

Format name that the backend needs:

class dmr.throttling.backends.redis.AsyncRedis(client: aioredis.Redis[Any])

Uses async Redis client for multiproccess safe rate-limiting.

This backend requires redis client library to be installed. We don’t ship it with the django-modern-rest.

You would have to install it separately:

pip install redis

See also

https://redis.readthedocs.io

async get(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str) → CachedRateLimit | None

Async get the cached rate limit state.

async incr(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: AsyncThrottle, *, cache_key: str, algorithm: BaseThrottleAlgorithm) → CachedRateLimit

Async increment cached rate limit state.

initialize_algorithm(algorithm: BaseThrottleAlgorithm) → None

Initialize and prepare backend for the algorithm.

needs_transaction_script: ClassVar[str] = 'lua'

Format name that the backend needs:

Algorithms

class dmr.throttling.algorithms.BaseThrottleAlgorithm

Base class for all throttling algorithms.

abstractmethod access(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → CachedRateLimit

Called when new access attempt is made.

Returns:

Cached rate limiting state.

Raises:

dmr.exceptions.TooManyRequestsError – when the limit is overused.

abstractmethod report_usage(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → dict[str, str]

Reports the throttling usage, but does not additionally increment.

transaction_script(script_format: str) → str | None

Optionally dump the transaction script for backends that support it.

It can be .lua or .sql.

class dmr.throttling.algorithms.SimpleRate

Simple rate algorithm.

Defines a fixed window with a fixed amount of requests possible. When window is expired, resets the count of requests.

access(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → CachedRateLimit

Check access.

report_usage(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → dict[str, str]

Reports the throttling usage, but does not additionally increment.

transaction_script(script_format: str) → str | None

Optionally dump the transaction script for backends that support it.

It can be .lua or .sql.

class dmr.throttling.algorithms.LeakyBucket

Leaky bucket algorithm.

Requests fill the bucket; tokens leak at a steady rate. Unlike dmr.throttling.algorithms.SimpleRate, which resets after a fixed window, LeakyBucket drains continuously providing smoother rate-limiting without allowing bursts at window boundaries.

Internally, the bucket request level is stored in scaled units as level * duration so all arithmetic stays integer-only. Each request adds duration scaled units to the level. Every elapsed second max_requests scaled units leak out.

See also

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

access(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → CachedRateLimit

Check access; raise when the bucket is full.

report_usage(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit | None) → dict[str, str]

Report throttling usage without incrementing.

transaction_script(script_format: str) → str | None

Optionally dump the transaction script for backends that support it.

It can be .lua or .sql.

Cache keys

class dmr.throttling.cache_keys.BaseThrottleCacheKey(*, runs_before_auth: bool, name: str)

Base class for all cache keys.

abstractmethod __call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Returns the cache key.

If string is returned, we use this as a cache key. If None is returned, this request will be skipped from this exact throttling check. However, other keys may still be applied.

class dmr.throttling.cache_keys.RemoteAddr(*, runs_before_auth: bool = True, name: str = 'RemoteAddr')

Uses REMOTE_ADDR from request.META as a cache key.

Warning

Be sure to correctly configure your HTTP Proxy! Otherwise, REMOTE_ADDR might be set incorrectly.

See also

See django.http.HttpRequest.META docs.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Return REMOTE_ADDR which is a user’s IP address, if it exists.

class dmr.throttling.cache_keys.UserPk(*, runs_before_auth: Literal[False] = False, name: str = 'UserPk', exclude_superuser: bool = True, exclude_stuff: bool = True)

Uses request.user.pk as a cache key.

Returns None for users that should be excluded from throttling checks or when pk is not set.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Return request.user.pk when user should be throttled.

class dmr.throttling.cache_keys.JwtToken(*, runs_before_auth: Literal[False] = False, name: str = 'JwtToken')

Uses a hash of JWT claims from request.__dmr_jwt__ as a cache key.

1. Never use a full token string for cache key generation.

2. Prefer jti claim, fallback to sub claim.

3. Store only SHA-256 hash, never raw claim values.

Returns None when jwt token is not set, or when both jti and sub are missing.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → str | None

Return a hash of JWT jti / sub claims as a cache key.

Headers

class dmr.throttling.headers.BaseResponseHeadersProvider

Base class for all header providers.

abstractmethod provide_headers_specs() → dict[str, HeaderSpec]

Provide a spec for headers for the OpenAPI.

abstractmethod response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: _BaseThrottle[Any], remaining: int, reset: int, *, report_all: bool = True) → dict[str, str]

Return a dict of rendered headers to be added to the response.

class dmr.throttling.headers.XRateLimit

Provides X-RateLimit headers.

There headers inside:

• X-RateLimit-Limit

• X-RateLimit-Remaining

• X-RateLimit-Reset

It is based on popular convention and does not have a formal spec.

See also

https://http.dev/x-ratelimit-limit

provide_headers_specs() → dict[str, HeaderSpec]

Provides headers specification.

response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: _BaseThrottle[Any], remaining: int, reset: int, *, report_all: bool = True) → dict[str, str]

Returns the formatted response headers.

class dmr.throttling.headers.RetryAfter

Provides Retry-After header.

It is based on the existing spec.

See also

RFC-6585 and RFC-7231

provide_headers_specs() → dict[str, HeaderSpec]

Provides headers specification.

response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: _BaseThrottle[Any], remaining: int, reset: int, *, report_all: bool = True) → dict[str, str]

Returns the formatted response headers.

class dmr.throttling.headers.RateLimitIETFDraft

Provides RateLimit and RateLimit-Policy headers.

It is based on the latest draft of the ratelmiting headers spec.

See also

https://www.ietf.org/archive/id/draft-ietf-httpapi-ratelimit-headers-10.html

provide_headers_specs() → dict[str, HeaderSpec]

Provides headers specification.

response_headers(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: _BaseThrottle[Any], remaining: int, reset: int, *, report_all: bool = True) → dict[str, str]

Returns the formatted response headers.