Flask-Limiter

Flask-Limiter provides rate limiting features to Flask applications.

By adding the extension to your flask application, you can configure various rate limits at different levels (e.g. application wide, per Blueprint, routes, resource etc).

Flask-Limiter can be configured to persist the rate limit state to many commonly used storage backends via the limits library.

Let’s get started!

Installation

Flask-Limiter can be installed via pip.

$ pip install Flask-Limiter

To include extra dependencies for a specific storage backend you can add the specific backend name via the extras notation. For example:

Redis

$ pip install Flask-Limiter[redis]

Memcached

$ pip install Flask-Limiter[memcached]

MongoDB

$ pip install Flask-Limiter[mongodb]

Quick start

A very basic setup can be achieved as follows:

from flask import Flask
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)
limiter = Limiter(
    app,
    key_func=get_remote_address,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="memory://",
)
@app.route("/slow")
@limiter.limit("1 per day")
def slow():
    return ":("

@app.route("/medium")
@limiter.limit("1/second", override_defaults=False)
def medium():
    return ":|"

@app.route("/fast")
def fast():
    return ":)"

@app.route("/ping")
@limiter.exempt
def ping():
    return "PONG"

The above Flask app will have the following rate limiting characteristics:

• Use an in-memory storage provided by limits.storage.MemoryStorage.

  Note

  This is only meant for testing/development and should be replaced with an appropriate storage of your choice before moving to production.

• Rate limiting by the remote_address of the request

• A default rate limit of 200 per day, and 50 per hour applied to all routes.

• The slow route having an explicit rate limit decorator will bypass the default rate limit and only allow 1 request per day.

• The medium route inherits the default limits and adds on a decorated limit of 1 request per second.

• The ping route will be exempt from any default rate limits.

  Tip

  The built in flask static files routes are also exempt from rate limits.

Every time a request exceeds the rate limit, the view function will not get called and instead a 429 http error will be raised.

The Flask-Limiter extension

The extension can be initialized with the flask.Flask application in the usual ways.

Using the constructor

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(app, key_func=get_remote_address)

Deferred app initialization using init_app()

limiter = Limiter(key_func=get_remote_address)
limiter.init_app(app)

At this point it might be a good idea to look at the configuration options available in the extension in the Using Flask Config section and the flask_limiter.Limiter class documentation.

Configuring a storage backend

The extension can be configured to use any storage supported by limits. Here are a few common examples:

Memcached

Any additional parameters provided in storage_options will be passed to the constructor of the memcached client (either PooledClient or HashClient). For more details see MemcachedStorage.

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(
    app,
    key_func=get_remote_address,
    storage_uri="memcached://localhost:11211",
    storage_options={}
)

Redis

Any additional parameters provided in storage_options will be passed to redis.Redis.from_url() as keyword arguments. For more details see RedisStorage

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(
  app, key_func=get_remote_address,
  storage_uri="redis://localhost:6379",
  storage_options={"connect_timeout": 30},
  strategy="fixed-window", # or "moving-window"
)

Redis (reused connection pool)

If you wish to reuse a redis.connection.ConnectionPool instance you can pass that in storage_option

import redis
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

pool = redis.connection.BlockingConnectionPool.from_url("redis://.....")
limiter = Limiter(
  app, key_func=get_remote_address,
  storage_uri="redis://",
  storage_options={"connection_pool": pool},
  strategy="fixed-window", # or "moving-window"
)

Redis Cluster

Any additional parameters provided in storage_options will be passed to RedisCluster as keyword arguments. For more details see RedisClusterStorage

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(
  app,
  key_func=get_remote_address,
  storage_uri="redis+cluster://localhost:7000,localhost:7001,localhost:7002",
  storage_options={"connect_timeout": 30},
  strategy="fixed-window", # or "moving-window"
)

MongoDB

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
....

limiter = Limiter(
  app, key_func=get_remote_address,
  storage_uri="mongodb://localhost:27017",
  strategy="fixed-window", # or "moving-window"
)

The storage_uri and storage_options parameters can also be provided by Using Flask Config variables. The different configuration options for each storage can be found in the storage backend documentation for limits as that is delegated to the limits library.

Rate Limit Domain

Each Limiter instance must be initialized with a key_func that returns the bucket in which each request is put into when evaluating whether it is within the rate limit or not.

For simple setups a utility function is provided: get_remote_address() which uses the remote_addr from flask.Request.

Please refer to Deploying an application behind a proxy for an example.

Decorators to declare rate limits

Decorators made available as instance methods of the Limiter instance to be used with the flask.Flask application.

Limiter.limit(limit_value: Union[str, Callable[[], str]], key_func: Optional[Callable[[], str]] = None, per_method: bool = False, methods: Optional[List[str]] = None, error_message: Optional[str] = None, exempt_when: Optional[Callable[[], bool]] = None, override_defaults: bool = True, deduct_when: Optional[Callable[[Response], bool]] = None, on_breach: Optional[Callable[[RequestLimit], Optional[Response]]] = None, cost: Union[int, Callable[[], int]] = 1) → LimitDecorator

decorator to be used for rate limiting individual routes or blueprints.

Parameters:

• limit_value – rate limit string or a callable that returns a string. Rate limit string notation for more details.

• key_func – function/lambda to extract the unique identifier for the rate limit. defaults to remote address of the request.

• per_method – whether the limit is sub categorized into the http method of the request.

• methods – if specified, only the methods in this list will be rate limited (default: None).

• error_message – string (or callable that returns one) to override the error message used in the response.

• exempt_when – function/lambda used to decide if the rate limit should skipped.

• override_defaults –

  whether the decorated limit overrides the default limits (Default: True).

  Note

  When used with a Blueprint the meaning of the parameter extends to any parents the blueprint instance is registered under. For more details see Nested Blueprints

• deduct_when – a function that receives the current flask.Response object and returns True/False to decide if a deduction should be done from the rate limit

• on_breach – a function that will be called when this limit is breached. If the function returns an instance of flask.Response that will be the response embedded into the RateLimitExceeded exception raised.

• cost – The cost of a hit or a function that takes no parameters and returns the cost as an integer (Default: 1).

There are a few ways of using the limit() decorator depending on your preference and use-case.

Single decorator

The limit string can be a single limit or a delimiter separated string

@app.route("....")
@limiter.limit("100/day;10/hour;1/minute")
def my_route()
  ...

Multiple decorators

The limit string can be a single limit or a delimiter separated string or a combination of both.

@app.route("....")
@limiter.limit("100/day")
@limiter.limit("10/hour")
@limiter.limit("1/minute")
def my_route():
  ...

Custom keying function

By default rate limits are applied based on the key function that the Limiter instance was initialized with. You can implement your own function to retrieve the key to rate limit by when decorating individual routes. Take a look at Rate Limit Key Functions for some examples..

def my_key_func():
  ...

@app.route("...")
@limiter.limit("100/day", my_key_func)
def my_route():
  ...

Note

The key function is called from within a flask request context.

Dynamically loaded limit string(s)

There may be situations where the rate limits need to be retrieved from sources external to the code (database, remote api, etc…). This can be achieved by providing a callable to the decorator.

def rate_limit_from_config():
    return current_app.config.get("CUSTOM_LIMIT", "10/s")

@app.route("...")
@limiter.limit(rate_limit_from_config)
def my_route():
    ...

Warning

The provided callable will be called for every request on the decorated route. For expensive retrievals, consider caching the response.

Note

The callable is called from within a flask request context during the before_request phase.

Exemption conditions

Each limit can be exempted when given conditions are fulfilled. These conditions can be specified by supplying a callable as an exempt_when argument when defining the limit.

@app.route("/expensive")
@limiter.limit("100/day", exempt_when=lambda: current_user.is_admin)
def expensive_route():
  ...

Reusable limits

For scenarios where a rate limit should be shared by multiple routes (For example when you want to protect routes using the same resource with an umbrella rate limit).

Limiter.shared_limit(limit_value: Union[str, Callable[[], str]], scope: Union[str, Callable[[str], str]], key_func: Optional[Callable[[], str]] = None, error_message: Optional[str] = None, exempt_when: Optional[Callable[[], bool]] = None, override_defaults: bool = True, deduct_when: Optional[Callable[[Response], bool]] = None, on_breach: Optional[Callable[[RequestLimit], Optional[Response]]] = None, cost: Union[int, Callable[[], int]] = 1) → LimitDecorator

decorator to be applied to multiple routes sharing the same rate limit.

Parameters:

• limit_value – rate limit string or a callable that returns a string. Rate limit string notation for more details.

• scope – a string or callable that returns a string for defining the rate limiting scope.

• key_func – function/lambda to extract the unique identifier for the rate limit. defaults to remote address of the request.

• error_message – string (or callable that returns one) to override the error message used in the response.

• exempt_when (function) – function/lambda used to decide if the rate limit should skipped.

• override_defaults –

  whether the decorated limit overrides the default limits. (default: True)

  Note

  When used with a Blueprint the meaning of the parameter extends to any parents the blueprint instance is registered under. For more details see Nested Blueprints

• deduct_when – a function that receives the current flask.Response object and returns True/False to decide if a deduction should be done from the rate limit

• on_breach – a function that will be called when this limit is breached. If the function returns an instance of flask.Response that will be the response embedded into the RateLimitExceeded exception raised.

• cost – The cost of a hit or a function that takes no parameters and returns the cost as an integer (default: 1).

Named shared limit

mysql_limit = limiter.shared_limit("100/hour", scope="mysql")

@app.route("..")
@mysql_limit
def r1():
    ...

@app.route("..")
@mysql_limit
def r2():
    ...

Dynamic shared limit

When a callable is passed as scope, the return value of the function will be used as the scope. Note that the callable takes one argument: a string representing the request endpoint.

def host_scope(endpoint_name):
    return request.host
host_limit = limiter.shared_limit("100/hour", scope=host_scope)

@app.route("..")
@host_limit
def r1():
    ...

@app.route("..")
@host_limit
def r2():
    ...

Decorators for skipping rate limits

Limiter.exempt(obj: Blueprint, *, flags: ExemptionScope = ExemptionScope.APPLICATION | ExemptionScope.DEFAULT) → Blueprint

Limiter.exempt(obj: Callable[[...], R], *, flags: ExemptionScope = ExemptionScope.APPLICATION | ExemptionScope.DEFAULT) → Callable[[...], R]

Limiter.exempt(*, flags: ExemptionScope = ExemptionScope.APPLICATION | ExemptionScope.DEFAULT) → Union[Callable[[Callable[[P], R]], Callable[[P], R]], Callable[[Blueprint], Blueprint]]

Mark a view function or all views in a blueprint as exempt from rate limits.

Parameters:

• obj – view function or blueprint to mark as exempt.

• flags – Controls the scope of the exemption. By default application wide limits and defaults configured on the extension are opted out of. Additional flags can be used to control the behavior when obj is a Blueprint that is nested under another Blueprint or has other Blueprints nested under it (See Nested Blueprints)

The method can be used either as a decorator without any arguments (the default flags will apply and the route will be exempt from default and application limits:

@app.route("...")
@limiter.exempt
def route(...):
   ...

Specific exemption flags can be provided at decoration time:

@app.route("...")
@limiter.exempt(flags=ExemptionScope.APPLICATION)
def route(...):
    ...

If an entire blueprint (i.e. all routes under it) are to be exempted the method can be called with the blueprint as the first parameter and any additional flags:

bp = Blueprint(...)
limiter.exempt(bp)
limiter.exempt(
    bp,
    flags=ExemptionScope.DEFAULT|ExemptionScope.APPLICATION|ExemptionScope.ANCESTORS
)

This decorator marks a function as a filter for requests that are going to be tested for rate limits. If any of the request filters return True no rate limiting will be performed for that request. This mechanism can be used to create custom white lists.

Limiter.request_filter(fn: Callable[[], bool]) → Callable[[], bool]

decorator to mark a function as a filter to be executed to check if the request is exempt from rate limiting.

Parameters:

fn – The function will be called before evaluating any rate limits to decide whether to perform rate limit or skip it.

@limiter.request_filter
def header_whitelist():
    return request.headers.get("X-Internal", "") == "true"

@limiter.request_filter
def ip_whitelist():
    return request.remote_addr == "127.0.0.1"

In the above example, any request that contains the header X-Internal: true or originates from localhost will not be rate limited.

For more complex use cases, refer to the Recipes section.