Rate Limiting

Introduction

Rate limiting is useful for certain public API endpoints. Examples are:

  • Login endpoints - reducing the speed with which passwords can be brute forced.

  • Computationally intensive endpoints, which could lead to a DOS attack.


RateLimitingMiddleware

Usage is simple - just wrap your ASGI app:

from piccolo_api.rate_limiting.middleware import RateLimitingMiddleware

app = RateLimitingMiddleware(my_asgi_app)

Source

class piccolo_api.rate_limiting.middleware.RateLimitingMiddleware(app: ASGIApp, provider: RateLimitProvider | None = None)[source]

Blocks clients who exceed a given number of requests in a given time period.

Parameters:
  • app – The ASGI app to wrap.

  • provider – Provides the logic around rate limiting. If not specified, it will default to a InMemoryLimitProvider.


Providers

The middleware can work with different Providers, which are responsible for storing traffic data, and signalling when a rate limit has been exceeded.

By default InMemoryLimitProvider is used.


InMemoryLimitProvider

Stores the traffic data in memory. You can customise it as follows:

app = RateLimitingMiddleware(
    my_asgi_app,
    provider=InMemoryLimitProvider(
        limit=1000,
        timespan=300
    ),
)

The limit is the number of requests needed by a client within the timespan (measured in seconds) to trigger a 429 error (too many requests).

If you want a blocked client to be allowed access again after a time period, specify this using the block_duration argument:

app = RateLimitingMiddleware(
    my_asgi_app,
    provider=InMemoryLimitProvider(
        limit=1000,
        timespan=300,
        block_duration=300  # Blocked for 5 minutes
    ),
)

Source

class piccolo_api.rate_limiting.middleware.InMemoryLimitProvider(timespan: int, limit: int = 1000, block_duration: int | None = None)[source]

A very simple rate limiting provider - works fine when running a single application instance.

Time values are given in seconds, rather than a timedelta, for improved performance.

Parameters:
  • timespan – The time in seconds between resetting the number of requests. Beware setting it too high, because memory usage will increase.

  • limit – The number of requests in the timespan, before getting blocked.

  • block_duration – If set, the number of seconds before a client is no longer blocked. Otherwise, they’re only removed when the app is restarted.


Custom Providers

Making a provider is simple, if the built-in ones don’t meet your needs. A provider needs to implement a simple interface - see RateLimitProvider.

Source

class piccolo_api.rate_limiting.middleware.RateLimitProvider[source]

An abstract base class which all rate limit providers should inherit from.

abstract increment(identifier: str)[source]
Parameters:

identifier – A unique identifier for the client (for example, IP address).

Raises:

RateLimitError – If too many requests are received from a client with this identifier.