Drop-in rate limiter for FastAPI, Django, Flask and any
Python backend.
A couple of atomic operations per request โ no Redis, no network hop. The hot path runs in
Rust and releases the GIL.
pip install rust-py-rate-limit
Rust handles the counting and concurrency. Python stays simple.
Up to limit requests per window_seconds per key. Predictable, cheap, and easy
to reason about โ the counter resets when the window rolls over.
Rate-limit by IP, user id, API key, route โ anything you can express as a string.
allow("ip:1.2.3.4") and you're done.
State lives in a sharded DashMap; stats use AtomicU64. No global lock on the
critical path โ different keys never contend.
Add RateLimitMiddleware in one line. Sets X-RateLimit-* and
Retry-After headers, returns 429 when over the limit.
Drop RateLimitMiddleware into MIDDLEWARE and configure it from
settings.py. Limit by IP or authenticated user.
Wrap any callable with @limiter.limit("login"). Raises RateLimitExceeded โ
with a dynamic key option for per-argument budgets.
stats() returns allowed, blocked, total_checks and
active_keys โ ready for dashboards and alerts.
The counting logic compiles to a native .so extension. No Rust toolchain needed to use it โ
just pip install.
cleanup_expired() reclaims memory from keys whose window has passed โ call it on a timer or
leave it to grow with your key space.
One middleware line, a decorator, or call the core API directly.
from rust_py_rate_limit import RateLimiter
limiter = RateLimiter(limit=3, window_seconds=60)
limiter.allow("ip:127.0.0.1") # True
limiter.allow("ip:127.0.0.1") # True
limiter.allow("ip:127.0.0.1") # True
limiter.allow("ip:127.0.0.1") # False โ limit reached
limiter.check("ip:127.0.0.1")
# {"allowed": False, "limit": 3, "remaining": 0,
# "reset_after_seconds": 42, "retry_after_seconds": 42}
limiter.remaining("ip:127.0.0.1") # 0
limiter.reset("ip:127.0.0.1") # True
from fastapi import FastAPI
from rust_py_rate_limit.fastapi import RateLimitMiddleware
app = FastAPI()
# 429 + X-RateLimit-* headers, automatically
app.add_middleware(
RateLimitMiddleware,
limit=100,
window_seconds=60,
key_func=lambda request: request.client.host,
)
@app.get("/api/users")
def list_users():
return {"users": []}
# settings.py
MIDDLEWARE = [
"rust_py_rate_limit.django.RateLimitMiddleware",
# ... rest of your middleware
]
RUST_PY_RATE_LIMIT = {
"LIMIT": 100,
"WINDOW_SECONDS": 60,
"KEY": "ip", # "ip" or "user"
}
from rust_py_rate_limit import RateLimiter, RateLimitExceeded
limiter = RateLimiter(limit=5, window_seconds=60)
@limiter.limit("login")
def login():
return "ok"
# Dynamic key derived from the function arguments
@limiter.limit(lambda user_id: f"user:{user_id}")
def fetch(user_id):
return user_id
# Raises RateLimitExceeded once the limit is exceeded
Each key gets a counter and a window. Cross the limit and you're blocked until it rolls over.
allow(key) consume โ boolcheck(key) consume โ dictremaining(key) peek, no consumereset(key) drop one keyclear() drop everythingcleanup_expired() reclaim memoryDashMap sharded, per-key locksAtomicU64 lock-free statsallow_threads releases the GIL{
"allowed": True,
"limit": 100,
"remaining": 99,
"reset_after_seconds": 60,
"retry_after_seconds": 0
}
{
"allowed": False,
"limit": 100,
"remaining": 0,
"reset_after_seconds": 42,
"retry_after_seconds": 42
}
Rust counts and stores. Python integrates. You stay fast.
The native .so extension is compiled once at publish time via maturin +
PyO3.
Your users just pip install โ no Rust toolchain required. State is per-process: behind multiple
workers each worker keeps its own counters.
Install from PyPI. No Rust, no compilers, no configuration.