philiprehberger-rate-limiter 0.5.0

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

philiprehberger-rate-limiter

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

Installation

pip install philiprehberger-rate-limiter

Usage

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

from philiprehberger_rate_limiter import RateLimiter

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

Usage Statistics

from philiprehberger_rate_limiter import RateLimiter

limiter = RateLimiter(100, 60)
limiter.allow("user-123")

stats = limiter.get_stats("user-123")
print(f"Current usage: {stats.current_usage}")
print(f"Remaining: {stats.remaining}")
print(f"Reset at: {stats.reset_at}")

Format status

from philiprehberger_rate_limiter import RateLimiter

limiter = RateLimiter(100, 60)
for _ in range(15):
    limiter.allow("user:123")

print(limiter.format_status("user:123"))
# "15/100 requests used (85 remaining); resets in 23.5s"

Blocking Wait

from philiprehberger_rate_limiter import RateLimiter

limiter = RateLimiter(100, 60)

# 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")

Async Context Manager

from philiprehberger_rate_limiter import RateLimiter

limiter = RateLimiter(100, 60)

async with limiter:
    status = await limiter.async_acquire("user-123")
    if status.allowed:
        await handle_request()

Decorator

from philiprehberger_rate_limiter import RateLimiter, rate_limit

# Standalone decorator
@rate_limit(calls=10, period=60)
def api_endpoint():
    return {"data": "ok"}

# Instance-based decorator
limiter = RateLimiter(10, 60)

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

# Async function decorator
@rate_limit(calls=10, period=60)
async def async_endpoint():
    return {"data": "ok"}

Rate Limiter Groups

from philiprehberger_rate_limiter import RateLimiter, RateLimiterGroup

limiter = RateLimiter(100, 60)
group = RateLimiterGroup(limiter, ["api-key-1", "api-key-2", "api-key-3"])

# All keys in the group share one rate limit pool
group.allow("api-key-1")  # consumes from shared pool
group.allow("api-key-2")  # consumes from same pool

stats = group.get_stats()
print(f"Group usage: {stats.current_usage}/{stats.limit}")

Algorithms

from philiprehberger_rate_limiter import RateLimiter, Algorithm

# 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)

# Leaky bucket — constant drain rate, smooths bursts
RateLimiter(100, 60, Algorithm.LEAKY_BUCKET)

API

Function / Class	Description
RateLimiter(requests, window_seconds, algorithm)	Create a rate limiter
limiter.allow(key)	Check if request is allowed
limiter.status(key)	Get detailed LimitStatus
limiter.get_stats(key)	Get RateLimiterStats without consuming a request
limiter.format_status(key)	Get a human-readable single-line summary (e.g. "15/100 requests used (85 remaining); resets in 23.5s")
limiter.wait(key, timeout)	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")
RateLimiterGroup(limiter, keys)	Create a shared rate limit group
group.allow(key)	Check if request is allowed against shared pool
group.status(key)	Get shared LimitStatus
group.get_stats()	Get shared RateLimiterStats
group.reset()	Reset shared group state
rate_limit(calls, period, algorithm)	Standalone decorator for rate limiting
format_status(status)	Format a LimitStatus as a human-readable summary string
Algorithm	Enum: FIXED_WINDOW, SLIDING_WINDOW, TOKEN_BUCKET, LEAKY_BUCKET
LimitStatus	Dataclass: allowed, remaining, reset_at, limit
RateLimiterStats	Dataclass: current_usage, remaining, reset_at, limit
RateLimitExceeded	Exception raised when rate limit is exceeded

Development

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

Support

If you find this project useful:

⭐ Star the repo

🐛 Report issues

💡 Suggest features

❤️ Sponsor development

🌐 All Open Source Projects

💻 GitHub Profile

🔗 LinkedIn Profile

License

MIT