fluxlimit 0.1.1

Distributed rate limiting with fairness for Python - token bucket, leaky bucket, and window algorithms with Redis and PostgreSQL backends.

fluxlimit

Distributed rate limiting with fairness for Python.

Production-ready rate limiting supporting token bucket, leaky bucket, fixed window, and sliding window algorithms — with strict distributed fairness via Redis Lua scripts or PostgreSQL advisory locks.

Why fluxlimit?

Feature	slowapi	fastapi-limiter	fluxlimit
Token bucket	❌	❌	✅
Leaky bucket	❌	❌	✅
Sliding window	❌	❌	✅
Distributed fairness	⚠️ Best-effort	⚠️ Best-effort	✅ Strict
Cost-based limiting	❌	❌	✅
PostgreSQL backend	❌	❌	✅
Multiple rules per request	❌	❌	✅

Quick Start

from fluxlimit import (
    RateLimiter,
    LimitRule,
    TokenBucket,
    TokenBucketConfig,
    MemoryBackend,
)

limiter = RateLimiter(
    backend=MemoryBackend(),
    default_rule=LimitRule(
        key="api",
        algorithm=TokenBucket(),
        config=TokenBucketConfig(key="api", rate=10, burst=20),
    ),
)

async with limiter:
    result = await limiter.check("user_123")
    print(result.allowed, result.remaining)

FastAPI Integration

FastAPI middleware is included in the core install:

pip install fluxlimit

from fastapi import FastAPI, Request
from fluxlimit import (
    RateLimiter,
    RateLimitMiddleware,
    LimitRule,
    TokenBucket,
    TokenBucketConfig,
    RedisBackend,
    rate_limit,
)

app = FastAPI()

limiter = RateLimiter(
    backend=RedisBackend("redis://localhost:6379"),
    default_rule=LimitRule(
        key="api",
        algorithm=TokenBucket(),
        config=TokenBucketConfig(key="api", rate=100, burst=150),
    ),
)

app.add_middleware(
    RateLimitMiddleware,
    limiter=limiter,
    skip_paths=["/health", "/health/*"],
)

@app.on_event("startup")
async def startup():
    await limiter.connect()

@app.on_event("shutdown")
async def shutdown():
    await limiter.disconnect()

@app.get("/items")
async def get_items():
    return {"message": "Hello World"}

ASGI middleware lives in fluxlimit.middleware.asgi. You can also import explicitly:

from fluxlimit.middleware.asgi import RateLimitMiddleware, rate_limit

Flask Integration

Install the Flask extra:

pip install fluxlimit[flask]

from flask import Flask
from fluxlimit import (
    RateLimiter,
    LimitRule,
    TokenBucket,
    TokenBucketConfig,
    MemoryBackend,
    Fluxlimit,
)

app = Flask(__name__)

limiter = RateLimiter(
    backend=MemoryBackend(),
    default_rule=LimitRule(
        key="api",
        algorithm=TokenBucket(),
        config=TokenBucketConfig(key="api", rate=10, burst=20),
    ),
)

fluxlimit = Fluxlimit(app, limiter=limiter, skip_paths=["/health", "/health/*"])

@app.route("/")
@fluxlimit.limit()
def index():
    return {"message": "Hello"}

@app.route("/health")
@fluxlimit.exempt
def health():
    return {"status": "ok"}

Flask integration lives in fluxlimit.middleware.flask.

Cost-Based Limiting

Different endpoints can consume different amounts of quota:

from fluxlimit import rate_limit

# Expensive endpoint costs 5 tokens
@app.post("/analyze")
@rate_limit(limiter, cost=5)
async def analyze(request: Request):
    return {"result": "..."}

Algorithms

Token Bucket

Allows controlled bursts while maintaining average rate. Best for APIs that need burst tolerance.

from fluxlimit import TokenBucket, TokenBucketConfig

TokenBucketConfig(key="api", rate=10, burst=20)  # 10/sec sustained, 20 burst

Leaky Bucket

Strict rate enforcement with no bursts. Best for protecting downstream services.

from fluxlimit import LeakyBucket, LeakyBucketConfig

LeakyBucketConfig(key="api", rate=10, capacity=100)  # 10/sec, queue of 100

Fixed Window

Simple counter in fixed time windows. Low memory, but allows bursts at boundaries.

from fluxlimit import FixedWindow, FixedWindowConfig

FixedWindowConfig(key="api", max_requests=100, window_seconds=60)

Sliding Window

Smooth rate limiting with no boundary bursts. Higher accuracy, more storage.

from fluxlimit import SlidingWindow, SlidingWindowConfig

SlidingWindowConfig(key="api", max_requests=100, window_seconds=60)

Backends

Backend	Use Case	Distributed Fairness
MemoryBackend	Single process, development	N/A
RedisBackend	Production, high throughput	✅ Lua scripts
PostgresBackend	Teams using PostgreSQL	✅ Advisory locks

PostgreSQL Setup

CREATE TABLE fluxlimit_rate_limits (
    key TEXT PRIMARY KEY,
    tokens FLOAT NOT NULL DEFAULT 0,
    volume FLOAT NOT NULL DEFAULT 0,
    count INTEGER NOT NULL DEFAULT 0,
    window_start FLOAT NOT NULL DEFAULT 0,
    last_update FLOAT NOT NULL DEFAULT 0,
    algorithm VARCHAR(32) NOT NULL DEFAULT 'token_bucket'
);

Installation

# Core (in-memory backend + FastAPI middleware)
pip install fluxlimit

# With Redis support
pip install fluxlimit[redis]

# With PostgreSQL support
pip install fluxlimit[postgres]

# With Prometheus metrics
pip install fluxlimit[prometheus]

# With Flask extension
pip install fluxlimit[flask]

# Everything
pip install fluxlimit[all]

Testing

Install development dependencies from the project root:

pip install ".[dev]"

Run the full test suite:

pytest

Run with verbose output or target a specific file:

pytest -v
pytest tests/test_core.py
pytest tests/test_algorithms.py::TestTokenBucket::test_refills_over_time

Generate a coverage report:

pytest --cov=fluxlimit --cov-report=term-missing

If you have not installed the package and only want to run tests against the source tree:

PYTHONPATH=src pytest

Test layout

File	Covers
tests/test_core.py	RateLimiter — allow/deny checks, check_or_raise, named rules
tests/test_algorithms.py	Token bucket burst/refill behavior, fixed window reset
tests/test_fastapi.py	ASGI middleware, @rate_limit, prefix skip, lazy imports
tests/test_flux_flask.py	Flask extension — decorators, exempt routes, skip paths
tests/test_middleware_imports.py	Lazy loading of middleware without eager framework imports

Tests use pytest-asyncio with asyncio_mode = auto (configured in pyproject.toml). The in-memory backend is used by default, so no Redis or PostgreSQL instance is required.

License

MIT