A lightweight, decorator-based rate limiter for Python applications.

rateflow protects synchronous functions, asynchronous functions, Flask routes, and FastAPI endpoints with four algorithms:

• Fixed window
• Sliding window
• Token bucket
• Leaky bucket

It supports process-local in-memory state and Redis-backed shared state.

Requirements

• Python 3.10 or newer
• Redis 6 or newer when using Redis storage

The current package release includes the Redis, Flask, FastAPI, Uvicorn, and pytest dependencies declared in the project metadata.

Installation

python -m pip install rateflow

Quick start

Configure a storage backend before defining decorated functions:

from rateflow import Algorithm, Configure, Storage, rate_limit
from rateflow.exceptions import RateLimitExceed

Configure.configure(Storage.MEMORY)


@rate_limit({
    "algorithm": Algorithm.FIXED_WINDOW,
    "calls": 5,
    "period": 60,
})
def get_data():
    return {"status": "ok"}


try:
    get_data()
except RateLimitExceed:
    print("rate limit exceeded")

The limiter consumes a permit before calling the wrapped function. A failed wrapped function still consumes the permit.

Configuration

The algorithm field is required. Required fields depend on the algorithm:

Algorithm	Required fields	Description
Algorithm.FIXED_WINDOW	calls, period	Allows a fixed number of calls during each period.
Algorithm.SLIDING_WINDOW	calls, period	Tracks permitted request timestamps over a rolling period.
Algorithm.TOKEN_BUCKET	capacity, refill_rate	Consumes tokens and replenishes them continuously.
Algorithm.LEAKY_BUCKET	capacity, leak_rate	Accepts requests while the bucket has capacity and drains continuously.

Invalid or incomplete configuration raises ValueError. The legacy refill_bucket field is accepted as an alias for refill_rate.

Token bucket

@rate_limit({
    "algorithm": Algorithm.TOKEN_BUCKET,
    "capacity": 20,
    "refill_rate": 2,  # tokens per second
})
def send_request():
    return {"status": "sent"}

Token buckets start full, refill continuously, and consume one token per permitted request.

Keys and shared limits

By default, each decorated function uses its qualified name as its storage key. A custom key can be supplied in the configuration or with the decorator's keyword argument:

@rate_limit(
    {"algorithm": Algorithm.FIXED_WINDOW, "calls": 10, "period": 60},
    key="api:search",
)
def search(query: str):
    return query

When both key forms are provided, the decorator argument takes precedence. Functions using the same key share rate-limit state, so keys should be stable and unique within the selected storage backend.

Storage backends

In-memory

Use in-memory storage for tests, local development, or a single-process application:

Configure.configure(Storage.MEMORY)

State is process-local and is lost when the process exits. If no backend is configured, the first decorator currently selects in-memory storage automatically; explicit configuration is recommended.

Redis

Use Redis when state must be shared between workers, containers, or hosts:

Configure.configure(
    Storage.REDIS,
    {
        "url": "redis://localhost:6379/0",
        "ttl": 3600,
        "socket_connect_timeout": 5,
    },
)

Redis must be running before application startup. Stored keys use a configurable TTL. Redis acquisition uses an optimistic transaction, but applications should still test their required concurrency guarantees under production load.

Async usage

The decorator supports coroutine functions and checks the limit before awaiting the endpoint:

from rateflow import Algorithm, Configure, Storage, rate_limit

Configure.configure(Storage.MEMORY)


@rate_limit({
    "algorithm": Algorithm.SLIDING_WINDOW,
    "calls": 10,
    "period": 60,
})
async def async_endpoint():
    return {"status": "ok"}

With Redis storage, the current check uses synchronous Redis I/O and may block the event loop during network or Redis delays.

Framework usage

The decorator can be applied to Flask and FastAPI endpoints in the same way as ordinary functions:

@app.get("/items")
@rate_limit({"algorithm": Algorithm.FIXED_WINDOW, "calls": 10, "period": 60})
def items():
    return {"status": "ok"}

Errors

• ValueError is raised for missing or invalid rate-limit configuration.
• RateLimitExceed is raised when a permitted request would exceed its limit.

License

MIT License: LICENSE