import time
from collections import defaultdict
import threading
 
 
class FixedWindowRateLimiter:
    """
    Fixed Window Counter Rate Limiter.
    Thread-safe implementation for single-process use.
    For distributed use, replace with Redis INCR + EXPIRE.
    """
 
    def __init__(self, limit: int, window_size_seconds: int):
        self.limit = limit
        self.window_size = window_size_seconds
        self.counters = defaultdict(int)
        self.lock = threading.Lock()
 
    def _get_window_key(self, identifier: str) -> str:
        current_window = int(time.time()) // self.window_size
        return f"{identifier}:{current_window}"
 
    def is_allowed(self, identifier: str) -> bool:
        """
        Returns True if the request is allowed, False if rate limited.
        """
        with self.lock:
            key = self._get_window_key(identifier)
            self.counters[key] += 1
            return self.counters[key] <= self.limit
 
    def get_remaining(self, identifier: str) -> int:
        with self.lock:
            key = self._get_window_key(identifier)
            used = self.counters.get(key, 0)
            return max(0, self.limit - used)
 
    def get_reset_time(self) -> int:
        """Returns Unix timestamp when the current window resets."""
        current_window = int(time.time()) // self.window_size
        return (current_window + 1) * self.window_size
 
 
# Usage
limiter = FixedWindowRateLimiter(limit=100, window_size_seconds=60)
 
for i in range(105):
    user_id = "user_123"
    if limiter.is_allowed(user_id):
        print(f"Request {i+1}: ALLOWED (remaining: {limiter.get_remaining(user_id)})")
    else:
        print(f"Request {i+1}: REJECTED (resets at: {limiter.get_reset_time()})")

import redis
import time
 
 
def is_allowed_fixed_window(
    redis_client: redis.Redis,
    identifier: str,
    limit: int,
    window_seconds: int
) -> tuple[bool, dict]:
    """
    Fixed window rate limiting using Redis INCR + EXPIRE.
    Returns (is_allowed, metadata_dict).
    """
    current_window = int(time.time()) // window_seconds
    key = f"rl:fw:{identifier}:{current_window}"
 
    # Pipeline for atomicity: INCR and EXPIRE in one round trip
    pipe = redis_client.pipeline()
    pipe.incr(key)
    pipe.ttl(key)
    results = pipe.execute()
 
    count = results[0]
    ttl = results[1]
 
    # Set expiry only on first request in this window
    if count == 1:
        redis_client.expire(key, window_seconds * 2)
        ttl = window_seconds
 
    allowed = count <= limit
    return allowed, {
        "limit": limit,
        "remaining": max(0, limit - count),
        "reset_in": ttl,
        "current_count": count
    }

import time
from collections import defaultdict, deque
import threading
 
 
class SlidingWindowLog:
    """
    Sliding Window Log Rate Limiter.
    Accurate but memory-intensive: O(limit) per user.
    """
 
    def __init__(self, limit: int, window_size_seconds: float):
        self.limit = limit
        self.window_size = window_size_seconds
        self.logs = defaultdict(deque)  # user_id -> deque of timestamps
        self.lock = threading.Lock()
 
    def is_allowed(self, identifier: str) -> tuple[bool, dict]:
        with self.lock:
            now = time.time()
            window_start = now - self.window_size
            log = self.logs[identifier]
 
            # Step 1: Evict timestamps outside the window
            while log and log[0] <= window_start:
                log.popleft()
 
            # Step 2: Count current requests in window
            current_count = len(log)
 
            # Step 3: Check against limit
            if current_count < self.limit:
                log.append(now)
                return True, {
                    "limit": self.limit,
                    "remaining": self.limit - current_count - 1,
                    "current_count": current_count + 1
                }
            else:
                # Calculate when the oldest request will expire
                retry_after = log[0] + self.window_size - now if log else 0
                return False, {
                    "limit": self.limit,
                    "remaining": 0,
                    "retry_after": retry_after,
                    "current_count": current_count
                }

import redis
import time
import uuid
 
 
def is_allowed_sliding_window_log(
    redis_client: redis.Redis,
    identifier: str,
    limit: int,
    window_seconds: int
) -> tuple[bool, dict]:
    """
    Sliding window log using Redis Sorted Sets (ZSET).
    Score = timestamp. Members = unique request IDs.
    """
    now = time.time()
    window_start = now - window_seconds
    key = f"rl:swl:{identifier}"
 
    # Use pipeline for batch execution
    pipe = redis_client.pipeline()
    # Remove timestamps outside the window
    pipe.zremrangebyscore(key, 0, window_start)
    # Count requests in window
    pipe.zcard(key)
    # Add current request with timestamp as score
    pipe.zadd(key, {str(uuid.uuid4()): now})
    # Set key expiry to auto-cleanup
    pipe.expire(key, window_seconds * 2)
    results = pipe.execute()
 
    count_before = results[1]  # count BEFORE adding this request
 
    if count_before < limit:
        return True, {
            "limit": limit,
            "remaining": limit - count_before - 1
        }
    else:
        # Remove the request we just added (it was rejected)
        # In production, use a Lua script to make this atomic
        redis_client.zpopmax(key)
        return False, {
            "limit": limit,
            "remaining": 0
        }

import time
from dataclasses import dataclass, field
from collections import defaultdict
import threading
 
 
@dataclass
class WindowState:
    prev_count: int = 0
    curr_count: int = 0
    curr_window_start: float = field(default_factory=time.time)
 
 
class SlidingWindowCounter:
    """
    Sliding Window Counter (Hybrid) Rate Limiter.
    Memory: O(1) per user (just two counters).
    Accuracy: Within 0.1% error for typical limits.
    Used by: Cloudflare, many production APIs.
    """
 
    def __init__(self, limit: int, window_size_seconds: float):
        self.limit = limit
        self.window_size = window_size_seconds
        self.states: dict[str, WindowState] = defaultdict(WindowState)
        self.lock = threading.Lock()
 
    def is_allowed(self, identifier: str) -> tuple[bool, dict]:
        with self.lock:
            now = time.time()
            state = self.states[identifier]
 
            # Check if we've moved past the current window
            elapsed = now - state.curr_window_start
 
            if elapsed >= self.window_size:
                # Advance window: current becomes previous
                state.prev_count = state.curr_count
                state.curr_count = 0
                state.curr_window_start = now
                elapsed = 0
 
            # What fraction of the previous window is still in the sliding range?
            # (window_size - elapsed) / window_size
            prev_weight = (self.window_size - elapsed) / self.window_size
 
            # Weighted estimate
            estimated_count = state.prev_count * prev_weight + state.curr_count
 
            if estimated_count < self.limit:
                state.curr_count += 1
                remaining = self.limit - int(estimated_count) - 1
                return True, {
                    "limit": self.limit,
                    "remaining": max(0, remaining),
                    "estimated_count": estimated_count + 1
                }
            else:
                return False, {
                    "limit": self.limit,
                    "remaining": 0,
                    "retry_after": self.window_size - elapsed
                }

import time
import threading
from dataclasses import dataclass
 
 
@dataclass
class BucketState:
    tokens: float
    last_refill: float
 
 
class TokenBucket:
    """
    Token Bucket Rate Limiter.
    Supports burst traffic while enforcing average rate.
    Used by: AWS API Gateway, Stripe, Nginx (burst param).
    """
 
    def __init__(self, capacity: float, refill_rate: float):
        """
        capacity: maximum tokens in the bucket (also max burst size)
        refill_rate: tokens added per second
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.states: dict[str, BucketState] = {}
        self.lock = threading.Lock()
 
    def _get_state(self, identifier: str) -> BucketState:
        if identifier not in self.states:
            self.states[identifier] = BucketState(
                tokens=self.capacity,
                last_refill=time.time()
            )
        return self.states[identifier]
 
    def _refill(self, state: BucketState, now: float) -> None:
        elapsed = now - state.last_refill
        tokens_to_add = elapsed * self.refill_rate
        state.tokens = min(self.capacity, state.tokens + tokens_to_add)
        state.last_refill = now
 
    def consume(self, identifier: str, tokens: float = 1.0) -> tuple[bool, dict]:
        """
        Attempt to consume 'tokens' from the bucket.
        Returns (allowed, metadata).
        """
        with self.lock:
            now = time.time()
            state = self._get_state(identifier)
            self._refill(state, now)
 
            if state.tokens >= tokens:
                state.tokens -= tokens
                wait_time = 0
                return True, {
                    "allowed": True,
                    "tokens_remaining": state.tokens,
                    "tokens_consumed": tokens
                }
            else:
                # How long until enough tokens are available?
                tokens_needed = tokens - state.tokens
                wait_time = tokens_needed / self.refill_rate
                return False, {
                    "allowed": False,
                    "tokens_remaining": state.tokens,
                    "retry_after_seconds": wait_time
                }
 
    def try_consume_or_wait(
        self, identifier: str,
        tokens: float = 1.0,
        max_wait_seconds: float = 0
    ) -> bool:
        """
        Optionally block/wait until a token is available.
        max_wait_seconds=0 means never wait (pure non-blocking).
        """
        allowed, meta = self.consume(identifier, tokens)
        if allowed:
            return True
 
        retry_after = meta.get("retry_after_seconds", 0)
        if max_wait_seconds > 0 and retry_after <= max_wait_seconds:
            time.sleep(retry_after)
            return self.try_consume_or_wait(identifier, tokens, 0)
        return False
 
 
# Usage Example
bucket = TokenBucket(capacity=10, refill_rate=2)  # 10 burst, 2/second sustained
 
# Fast burst - first 10 allowed
for i in range(12):
    allowed, meta = bucket.consume("user_123")
    print(f"Request {i+1}: {'ALLOW' if allowed else 'DENY'} "
          f"(tokens: {meta.get('tokens_remaining', 0):.1f})")
 
# Wait for tokens to refill
time.sleep(1)
print("\nAfter 1 second (2 tokens refilled):")
for i in range(3):
    allowed, meta = bucket.consume("user_123")
    print(f"Request {i+1}: {'ALLOW' if allowed else 'DENY'} "
          f"(tokens: {meta.get('tokens_remaining', 0):.1f})")

class CostBasedTokenBucket(TokenBucket):
    """
    Different operations cost different numbers of tokens.
    Example: GraphQL queries with varying complexity.
    """
    OPERATION_COSTS = {
        "simple_read":    1,
        "list_query":     3,
        "complex_search": 10,
        "export":         25,
        "bulk_delete":    50,
    }
 
    def is_operation_allowed(self, identifier: str, operation: str) -> bool:
        cost = self.OPERATION_COSTS.get(operation, 1)
        allowed, _ = self.consume(identifier, tokens=cost)
        return allowed

import queue
import threading
import time
from typing import Callable, Any
 
 
class LeakyBucket:
    """
    Leaky Bucket Rate Limiter.
    Output rate is always constant regardless of input bursts.
    Used for: Traffic shaping, smoothing output to downstream services.
    """
 
    def __init__(self, capacity: int, leak_rate: float):
        """
        capacity: max requests the queue can hold
        leak_rate: requests processed per second (constant)
        """
        self.capacity = capacity
        self.leak_rate = leak_rate
        self.queue = queue.Queue(maxsize=capacity)
        self.running = False
        self._processor_thread = None
 
    def submit(self, request: Any) -> bool:
        """
        Submit a request to the bucket.
        Returns True if accepted (queued), False if dropped (bucket full).
        """
        try:
            self.queue.put_nowait(request)
            return True
        except queue.Full:
            return False  # Drop the request
 
    def start(self, processor: Callable[[Any], None]) -> None:
        """Start the background processor that leaks at a fixed rate."""
        self.running = True
        self._processor_thread = threading.Thread(
            target=self._leak, args=(processor,), daemon=True
        )
        self._processor_thread.start()
 
    def _leak(self, processor: Callable[[Any], None]) -> None:
        interval = 1.0 / self.leak_rate
        while self.running:
            try:
                request = self.queue.get(timeout=interval)
                processor(request)
            except queue.Empty:
                pass
            time.sleep(interval)
 
    def stop(self) -> None:
        self.running = False
 
 
# Usage Example
def process_request(req):
    print(f"Processing: {req} at {time.time():.2f}")
 
bucket = LeakyBucket(capacity=20, leak_rate=5)  # queue 20, process 5/second
bucket.start(process_request)
 
# Send a burst of 30 requests instantly
for i in range(30):
    accepted = bucket.submit(f"request_{i+1}")
    if not accepted:
        print(f"request_{i+1}: DROPPED (bucket full)")
 
time.sleep(5)  # Watch them drain at 5/second
bucket.stop()

class LeakyBucketCounter:
    """
    Leaky bucket as a counter - useful for Redis implementation.
    Instead of a real queue, tracks the "water level" as a number.
    """
 
    def __init__(self, capacity: int, leak_rate: float):
        self.capacity = capacity
        self.leak_rate = leak_rate  # units per second
        self.levels: dict[str, dict] = {}
        self.lock = threading.Lock()
 
    def is_allowed(self, identifier: str) -> bool:
        with self.lock:
            now = time.time()
 
            if identifier not in self.levels:
                self.levels[identifier] = {
                    "level": 0.0,
                    "last_check": now
                }
 
            state = self.levels[identifier]
            elapsed = now - state["last_check"]
 
            # Leak out based on elapsed time
            leaked = elapsed * self.leak_rate
            state["level"] = max(0.0, state["level"] - leaked)
            state["last_check"] = now
 
            # Check if there's room
            if state["level"] + 1 <= self.capacity:
                state["level"] += 1
                return True
            return False

import time
from dataclasses import dataclass
import threading
 
 
class GCRARateLimiter:
    """
    GCRA (Generic Cell Rate Algorithm) Rate Limiter.
    Also known as: Virtual Scheduling Algorithm, Leaky Bucket as a Meter.
    Used by: Stripe's throttled library.
    Memory: O(1) per user (single timestamp).
    """
 
    def __init__(self, rate: float, burst: int):
        """
        rate: requests per second allowed (sustained)
        burst: maximum burst size above the sustained rate
        """
        self.emission_interval = 1.0 / rate  # seconds between requests at sustained rate
        self.burst_offset = burst * self.emission_interval  # TAT tolerance
        self.tat_store: dict[str, float] = {}  # user -> TAT
        self.lock = threading.Lock()
 
    def is_allowed(self, identifier: str) -> tuple[bool, dict]:
        with self.lock:
            now = time.time()
 
            # Get or initialize TAT
            previous_tat = self.tat_store.get(identifier, now)
 
            # New TAT = max(now, previous_TAT) + emission_interval
            new_tat = max(now, previous_tat) + self.emission_interval
 
            # Allowed if new TAT is within burst tolerance
            allowed_at = new_tat - self.burst_offset
 
            if allowed_at <= now:
                self.tat_store[identifier] = new_tat
                return True, {
                    "allowed": True,
                    "next_allowed_in": 0
                }
            else:
                # How long until the request would be allowed?
                wait_time = allowed_at - now
                return False, {
                    "allowed": False,
                    "retry_after_seconds": wait_time
                }
 
 
# Usage
gcra = GCRARateLimiter(rate=10, burst=5)  # 10 RPS sustained, 5 burst
 
# Test burst
for i in range(8):
    allowed, meta = gcra.is_allowed("user_123")
    print(f"Request {i+1}: {'ALLOW' if allowed else f'DENY (retry in {meta[\"retry_after_seconds\"]:.3f}s)'}")