philiprehberger-rate-limiter 0.3.0

In-memory rate limiter with sliding window and token bucket

philiprehberger-rate-limiter

In-memory rate limiter with sliding window and token bucket algorithms.

Installation

pip install philiprehberger-rate-limiter

Usage

Basic Rate Limiting

from philiprehberger_rate_limiter import RateLimiter, Algorithm

limiter = RateLimiter(
    requests=100,
    window_seconds=60,
    algorithm=Algorithm.SLIDING_WINDOW,
)

if limiter.allow("user-123"):
    handle_request()

Check Status

status = limiter.status("user-123")
print(f"Allowed: {status.allowed}")
print(f"Remaining: {status.remaining}/{status.limit}")
print(f"Resets at: {status.reset_at}")

Blocking Wait

# Synchronous — blocks until quota is available or timeout expires
status = limiter.wait("user-123", timeout=5.0)

# Async — awaits until quota is available
status = await limiter.async_acquire("user-123")

Decorator

limiter = RateLimiter(10, 60)

@limiter.limit("10/minute")
def api_endpoint():
    return {"data": "ok"}

Algorithms

# Fixed window — resets at interval boundaries
RateLimiter(100, 60, Algorithm.FIXED_WINDOW)

# Sliding window (default) — rolling time window
RateLimiter(100, 60, Algorithm.SLIDING_WINDOW)

# Token bucket — smooth rate with burst capacity
RateLimiter(100, 60, Algorithm.TOKEN_BUCKET)

API

Name	Description
RateLimiter(requests, window_seconds, algorithm=SLIDING_WINDOW)	Create a rate limiter
limiter.allow(key)	Check if request is allowed
limiter.status(key)	Get detailed LimitStatus
limiter.wait(key, timeout=10.0)	Block until quota available or raise RateLimitExceeded
limiter.async_acquire(key)	Async wait until quota is available
limiter.reset(key)	Reset state for a key
limiter.reset_all()	Reset state for all keys
limiter.active_keys()	List all keys with active state
limiter.limit(rate)	Decorator with rate string (e.g., "10/minute")

Development

pip install -e .
python -m pytest tests/ -v

Support

If you find this package useful, consider starring the repository.

License

MIT