rateflow

A lightweight, decorator-based rate limiter for Python applications

Protect synchronous functions, asynchronous functions, Flask routes, and FastAPI endpoints with fixed-window, sliding-window, token-bucket, or leaky-bucket algorithms.

---

Features

• Simple @rate_limit(...) decorator API.
• Mandatory configuration validation at decoration time.
• Synchronous and asynchronous function support.
• Four rate-limiting algorithms:
  • Fixed window
  • Sliding window
  • Token bucket
  • Leaky bucket
• In-memory storage for single-process applications and tests.
• Redis storage for shared state across processes and hosts.
• Framework examples for Flask and FastAPI.
• Local wheel and source-distribution workflow for running examples against the built package.

Requirements

• Python 3.10 or newer
• Redis 6 or newer when using Redis storage
• The current package metadata installs Redis, Flask, FastAPI, Uvicorn, and test tooling as unconditional dependencies.

Installation

Install from PyPI

Install the latest published release:

python -m pip install rateflow

PyPI page: pypi.org/project/rateflow

Install from a local build

To install a locally built wheel:

uv build
uv pip install --python .venv/bin/python --force-reinstall dist/rateflow-*.whl

The examples intentionally import rateflow from the installed package rather than directly from src.

Quick start

Initialize the storage backend once during application startup, before any decorated function is defined:

from rateflow import Algorithm, Configure, rate_limit, Storage

Configure.configure(Storage.MEMORY)


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


get_data()

After five permitted invocation attempts within the applicable 60-second fixed window, the decorator raises RateLimitExceed. The permit is consumed before the wrapped function runs, including when the function later raises an exception:

from rateflow.exceptions import RateLimitExceed

try:
	get_data()
except RateLimitExceed as exc:
	print(exc)

The default key is the decorated function's qualified name. Set a stable custom key when multiple functions should share one limit:

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

Keys and shared limits

Each decorated function uses its qualified name as the default storage key. A custom key can be supplied either in the configuration dictionary or with the decorator's key argument. When both are supplied, the decorator argument takes precedence. Functions using the same key share the same stored rate-limit state, so keys should be stable and unique within the selected storage backend.

Configuration

The algorithm field is required. The remaining required fields depend on the selected 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 individual request timestamps in a rolling period.
Algorithm.TOKEN_BUCKET	capacity, refill_rate	Consumes tokens and replenishes them continuously.
Algorithm.LEAKY_BUCKET	capacity, leak_rate	Enforces a bucket capacity while requests leak out over time.

Invalid or incomplete configuration raises ValueError before the decorated function is created. The legacy refill_bucket field is also accepted as an alias for refill_rate.

Algorithm behavior differs as follows:

• Fixed window resets the counter at fixed period boundaries.
• Sliding window tracks individual permitted request timestamps over a rolling period.
• Token bucket starts full, refills continuously, and consumes one token per permitted request.
• Leaky bucket continuously drains its level at leak_rate; it is not a fixed-window algorithm.

Fixed and sliding windows

@rate_limit({
	"algorithm": Algorithm.SLIDING_WINDOW,
	"calls": 100,
	"period": 60,
	"key": "api:read",
})
def read_resource():
	...

Token bucket

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

Leaky bucket

@rate_limit({
	"algorithm": Algorithm.LEAKY_BUCKET,
	"capacity": 20,
	"leak_rate": 2,  # requests per second
})
def process_job():
	...

Storage backends

In-memory storage

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

Configure.configure(Storage.MEMORY)

If no backend is configured, the first decorator currently selects the in-memory backend automatically. Configure the backend explicitly to avoid accidental process-local limiting.

State is process-local and is lost when the process exits.

Redis storage

Use Redis when rate-limit 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 the application starts. The Redis backend stores state with a configurable TTL so unused keys are eventually removed. Redis state is shared between workers, containers, and hosts, and acquisition uses an optimistic transaction to protect concurrent updates.

Async usage

The decorator preserves coroutine functions and performs the rate-limit check before awaiting the endpoint:

from rateflow import Algorithm, Configure, rate_limit, Storage

Configure.configure(Storage.MEMORY)


@rate_limit({
	"algorithm": Algorithm.TOKEN_BUCKET,
	"capacity": 10,
	"refill_rate": 1,
})
async def async_endpoint():
	return {"status": "ok"}

The async wrapper checks the limit before awaiting the endpoint. With the current Redis backend, this check uses synchronous Redis I/O and can block the event loop during network or Redis delays.

Flask

from flask import Flask, jsonify
from rateflow import Algorithm, Configure, rate_limit, Storage

Configure.configure(Storage.MEMORY)
app = Flask(__name__)


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

Run the included example from the repository root:

flask --app examples.flask_app run

FastAPI

from fastapi import FastAPI
from rateflow import Algorithm, Configure, rate_limit, Storage

Configure.configure(Storage.MEMORY)
app = FastAPI()


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

Run the included example from the repository root:

uvicorn examples.fastapi_app:app --reload

Included examples

File	Purpose
examples/inmemory_usage.py	All algorithms with in-memory storage.
examples/redis_usage.py	All algorithms with Redis storage.
examples/flask_app.py	Flask endpoints using in-memory storage.
examples/flask_redis.py	Flask endpoint using Redis storage.
examples/fastapi_app.py	Async FastAPI endpoints using in-memory storage.
examples/fastapi_redis.py	Async FastAPI endpoint using Redis storage.

See examples/README.md for the complete local-build workflow.

Testing

Run the test suite with:

python -m pytest -q

The tests cover all algorithms, key isolation, configuration validation, synchronous decorators, and asynchronous decorators. Redis integration examples require a reachable Redis server.

Project layout

src/rateflow/
├── algorithm/       # Rate-limiting algorithm implementations
├── models/          # Configuration, state, and result models
├── storage/         # In-memory and Redis backends
├── configure.py     # Process-wide storage initialization
├── decorators.py    # Sync and async decorator API
└── limiter.py       # Algorithm/storage orchestration
examples/            # Framework and storage usage examples
tests/               # Unit and integration-style tests

Error handling

• ValueError: missing or invalid algorithm configuration, unknown algorithms, or unknown storage backends.
• RateLimitExceed: raised when a request exceeds its configured limit.

License

This project is licensed under the MIT License.