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

Querying budget without consuming

from philiprehberger_rate_limiter import RateLimiter, Algorithm
import time

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

# How many requests still allowed (does not consume a request)
print(limiter.remaining("user-123"))  # 85

# Monotonic timestamp when capacity for one more request is available
reset = limiter.reset_at("user-123")
print(f"Available in {max(0.0, reset - time.monotonic()):.1f}s")

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

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