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: Tue, 14 Apr 2026 17:09:17 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: Tue, 14 Apr 2026 17:09:19 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"
                }
              }
            }
          }
        }
      }
    }
  }
}

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.DjangoCache as the backend. You can customize which 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 DjangoCache


class SyncController(Controller[PydanticSerializer]):
    throttling = (
        SyncThrottle(
            max_requests=1,
            duration_in_seconds=Rate.minute,
            backend=DjangoCache(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: Tue, 14 Apr 2026 17:09:20 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 or somewhere else. To do so, you would need to subclass dmr.throttling.backends.BaseThrottleBackend and override 4 methods.

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

• DjangoCache, default

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"

$ curl http://127.0.0.1:8000/api/sync/ -D - -X GET
HTTP/1.1 429 Too Many Requests
date: Tue, 14 Apr 2026 17:09:21 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 algorithms. To do so, you would need to subclass dmr.throttling.algorithms.BaseThrottleAlgorithm and override 3 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.

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

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: Tue, 14 Apr 2026 17:09:23 GMT
server: uvicorn
X-RateLimit-Limit: 1
X-RateLimit-Remaining: 0
X-RateLimit-Reset: 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/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: Tue, 14 Apr 2026 17:09:24 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: Tue, 14 Apr 2026 17:09:24 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: Tue, 14 Apr 2026 17:09:25 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: Tue, 14 Apr 2026 17:09:25 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: BaseThrottleBackend | None = None, algorithm: BaseThrottleAlgorithm | None = None, response_headers: Iterable[BaseResponseHeadersProvider] | None = None)

Sync throttle type for sync endpoints.

__call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → 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.

check(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str) → None

Check whether this request has rate limiting quota left.

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: BaseThrottleBackend | None = None, algorithm: BaseThrottleAlgorithm | None = None, response_headers: Iterable[BaseResponseHeadersProvider] | None = None)

Async throttle type for async endpoints.

async __call__(endpoint: Endpoint, controller: Controller[BaseSerializer]) → 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.

async check(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str) → None

Check whether this request has rate limiting quota left.

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.

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.BaseThrottleBackend

Base class for all throttling backends.

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

abstractmethod async aget(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str) → CachedRateLimit | None

Async get the cached rate limit state.

abstractmethod async aset(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str, cache_object: CachedRateLimit, *, ttl_seconds: int) → None

Async set the cached rate limit state.

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

Sync get the cached rate limit state.

abstractmethod set(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str, cache_object: CachedRateLimit, *, ttl_seconds: int) → None

Sync set the cached rate limit state.

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

Uses Django cache framework for storing the rate limiting state.

See also

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

__post_init__() → None

Initialize the cache backend.

async aget(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str) → CachedRateLimit | None

Async get the cached rate limit state.

async aset(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str, cache_object: CachedRateLimit, *, ttl_seconds: int) → None

Async set the cached rate limit state.

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

Sync get the cached rate limit state.

set(endpoint: Endpoint, controller: Controller[BaseSerializer], cache_key: str, cache_object: CachedRateLimit, *, ttl_seconds: int) → None

Sync set the cached rate limit state.

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 record(endpoint: Endpoint, controller: Controller[BaseSerializer], throttle: SyncThrottle | AsyncThrottle, cache_object: CachedRateLimit) → CachedRateLimit

Records successful access.

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.

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.

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

Record successful 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.

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.

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

Record access by adding request to the bucket.

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

Report throttling usage without incrementing.

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.

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, 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, 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, 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, remaining: int, reset: int, *, report_all: bool = True) → dict[str, str]

Returns the formatted response headers.