FixedWindow

Divides time into fixed-size buckets and tracks a request counter per bucket. A request is allowed while the counter is within limit; once the bucket rolls over the counter resets.

from core_redis.rate_limits import FixedWindow

limiter = FixedWindow(redis_kwargs={"host": "localhost", "port": 6379})
allowed = limiter.is_allowed("user_123", limit=100, window=60)

The counter is incremented and the TTL is set in a single Redis pipeline call (INCR + EXPIRE), keeping round-trips to one per request.

A common pattern is to guard outbound HTTP calls so a client never exceeds an upstream API’s rate limit:

import requests
from core_redis.rate_limits import FixedWindow

limiter = FixedWindow(redis_kwargs={"host": "localhost", "port": 6379})

def call_api(user_id: str) -> None:
    if not limiter.is_allowed(user_id, limit=100, window=60):
        print(f"[{user_id}] BLOCKED —> rate limit exceeded")
        return

    response = requests.get("https://api.example.com/data", timeout=5)
    print(f"[{user_id}] {response.status_code}")

SlidingWindowLog

Stores a timestamp for every request in a Redis sorted set. On each call, entries older than now − window are pruned before counting, so the window always reflects exactly the last window seconds, the burst problem does not occur.

Returns a (allowed, remaining) tuple so callers know how many slots are left without a second round-trip.

from core_redis.rate_limits import SlidingWindowLog

limiter = SlidingWindowLog(redis_kwargs={"host": "localhost", "port": 6379})
allowed, remaining = limiter.is_allowed("user_123", limit=100, window=60)

if not allowed:
    print("Rate limit exceeded")
else:
    print(f"{remaining} requests remaining in this window")

The same HTTP-guard pattern works here:

import requests
from core_redis.rate_limits import SlidingWindowLog

limiter = SlidingWindowLog(redis_kwargs={"host": "localhost", "port": 6379})

def call_api(user_id: str) -> None:
    allowed, remaining = limiter.is_allowed(user_id, limit=100, window=60)
    if not allowed:
        print(f"[{user_id}] BLOCKED —> rate limit exceeded")
        return

    response = requests.get("https://api.example.com/data", timeout=5)
    print(f"[{user_id}] {response.status_code}  ({remaining} remaining)")

TokenBucket

Maintains a virtual token bucket per identifier in a Redis hash. Tokens refill continuously at refill_rate per second up to capacity. Each request consumes tokens_per_request tokens. A request is allowed when the bucket has enough tokens; otherwise it is rejected.

Returns a (allowed, available_tokens) tuple.

from core_redis.rate_limits import TokenBucket

limiter = TokenBucket(redis_kwargs={"host": "localhost", "port": 6379})

allowed, tokens = limiter.is_allowed(
    "user_123",
    capacity=100,       # max burst size
    refill_rate=10.0,   # tokens added per second
)
if not allowed:
    print(f"Rate limited —> {tokens} tokens available")
else:
    print(f"Allowed —> {tokens} tokens remaining")

Variable-cost operations are supported via tokens_per_request:

# A bulk export costs 10 tokens; a lightweight read costs 1
allowed, tokens = limiter.is_allowed(
    "user_123", capacity=100, refill_rate=10.0, tokens_per_request=10
)

LeakyBucket

Maintains a virtual queue per identifier in a Redis hash. Incoming requests fill the queue; the queue drains at a fixed leak_rate requests per second regardless of arrival rate. A request is accepted when the queue has room; otherwise it is rejected immediately. Unlike TokenBucket, the output rate is strictly constant, bursts are absorbed into the queue and processed at the leak rate, never served faster.

Returns a (allowed, available) tuple where available is the number of free queue slots after this request (0 when blocked).

from core_redis.rate_limits import LeakyBucket

limiter = LeakyBucket(redis_kwargs={"host": "localhost", "port": 6379})

allowed, available = limiter.is_allowed(
    "user_123",
    capacity=100,     # max queue depth
    leak_rate=10.0,   # requests drained per second
)

if not allowed:
    print("Queue full - retry later")
else:
    print(f"Queued - {available} slots remaining")

HTTP-guard pattern:

import requests
from core_redis.rate_limits import LeakyBucket

limiter = LeakyBucket(redis_kwargs={"host": "localhost", "port": 6379})

def call_api(user_id: str) -> None:
    allowed, available = limiter.is_allowed(user_id, capacity=100, leak_rate=10.0)
    if not allowed:
        print(f"[{user_id}] BLOCKED - queue full")
        return

    response = requests.get("https://api.example.com/data", timeout=5)
    print(f"[{user_id}] {response.status_code}  ({available} slots remaining)")