# -*- coding: utf-8 -*- """ Sliding-Window-Algorithmus für Rate-Limiting. Implementiert das Sliding-Window-Verfahren für präzise zeitbasierte Durchsatzbegrenzung. """ from __future__ import annotations import threading import time from collections import deque from dataclasses import dataclass, field from datetime import datetime from typing import Any @dataclass class SlidingWindowStats: """ Statistiken für ein Sliding-Window. Bietet Einblick in die Nutzung des Zeitfensters. """ requests_allowed: int = 0 """Anzahl erlaubter Requests.""" requests_denied: int = 0 """Anzahl abgelehnter Requests.""" last_request: datetime | None = None """Zeitpunkt des letzten Requests.""" peak_requests_in_window: int = 0 """Maximale Requests in einem Fenster.""" @property def total_requests(self) -> int: """Gesamtzahl der Requests.""" return self.requests_allowed + self.requests_denied @property def allow_rate(self) -> float: """Erlaubnisrate in Prozent.""" if self.total_requests == 0: return 100.0 return (self.requests_allowed / self.total_requests) * 100 def record_allow(self, current_count: int) -> None: """Zeichnet einen erlaubten Request auf.""" self.requests_allowed += 1 self.last_request = datetime.now() self.peak_requests_in_window = max( self.peak_requests_in_window, current_count ) def record_deny(self) -> None: """Zeichnet einen abgelehnten Request auf.""" self.requests_denied += 1 self.last_request = datetime.now() def reset(self) -> None: """Setzt alle Statistiken zurück.""" self.requests_allowed = 0 self.requests_denied = 0 self.last_request = None self.peak_requests_in_window = 0 def to_dict(self) -> dict[str, Any]: """Konvertiert die Statistiken in ein Dictionary.""" return { "requests_allowed": self.requests_allowed, "requests_denied": self.requests_denied, "total_requests": self.total_requests, "allow_rate": self.allow_rate, "peak_requests_in_window": self.peak_requests_in_window, "last_request": ( self.last_request.isoformat() if self.last_request else None ), } class SlidingWindow: """ Sliding-Window-Implementierung für Rate-Limiting. Zählt Requests innerhalb eines gleitenden Zeitfensters. Im Gegensatz zum Token-Bucket bietet dies eine striktere Kontrolle ohne Burst-Möglichkeit. Attributes: window_size: Größe des Zeitfensters in Sekunden. max_requests: Maximale Requests im Fenster. Example: window = SlidingWindow(window_size=60.0, max_requests=100) if window.is_allowed(): # Request im Fenster erlaubt process_request() else: # Rate limit erreicht raise TooManyRequests() """ def __init__( self, window_size: float, max_requests: int, ) -> None: """ Initialisiert das Sliding-Window. Args: window_size: Fenstergröße in Sekunden. max_requests: Maximale Requests pro Fenster. """ self._window_size = window_size self._max_requests = max_requests self._timestamps: deque[float] = deque() self._lock = threading.Lock() self._stats = SlidingWindowStats() @property def window_size(self) -> float: """Gibt die Fenstergröße zurück.""" return self._window_size @property def max_requests(self) -> int: """Gibt das Maximum zurück.""" return self._max_requests @property def stats(self) -> SlidingWindowStats: """Gibt die Statistiken zurück.""" return self._stats def _cleanup(self, now: float) -> None: """Entfernt abgelaufene Einträge.""" cutoff = now - self._window_size while self._timestamps and self._timestamps[0] < cutoff: self._timestamps.popleft() @property def current_count(self) -> int: """Gibt die aktuelle Anzahl im Fenster zurück.""" with self._lock: self._cleanup(time.monotonic()) return len(self._timestamps) def is_allowed(self) -> bool: """ Prüft, ob ein Request erlaubt ist und zeichnet ihn auf. Returns: True wenn erlaubt. """ now = time.monotonic() with self._lock: self._cleanup(now) if len(self._timestamps) < self._max_requests: self._timestamps.append(now) self._stats.record_allow(len(self._timestamps)) return True self._stats.record_deny() return False def try_acquire(self, count: int = 1) -> tuple[bool, float]: """ Versucht, mehrere Slots zu reservieren. Args: count: Anzahl zu reservierender Slots. Returns: Tuple (erfolgreich, Wartezeit in Sekunden). """ now = time.monotonic() with self._lock: self._cleanup(now) available = self._max_requests - len(self._timestamps) if available >= count: for _ in range(count): self._timestamps.append(now) self._stats.record_allow(len(self._timestamps)) return True, 0.0 # Berechne Wartezeit bis genug Slots frei werden if self._timestamps: needed = count - available # Wann werden die nächsten 'needed' Slots frei? if len(self._timestamps) >= needed: oldest_needed = list(self._timestamps)[needed - 1] wait_time = (oldest_needed + self._window_size) - now wait_time = max(0, wait_time) else: wait_time = self._window_size else: wait_time = 0 self._stats.record_deny() return False, wait_time def wait_and_proceed( self, max_wait: float | None = None, ) -> bool: """ Wartet und reserviert dann einen Slot. Args: max_wait: Maximale Wartezeit in Sekunden. Returns: True wenn erfolgreich. """ start = time.monotonic() while True: success, wait_time = self.try_acquire() if success: return True if max_wait is not None: elapsed = time.monotonic() - start remaining = max_wait - elapsed if remaining <= 0: return False wait_time = min(wait_time, remaining) if wait_time > 0: time.sleep(wait_time) def remaining(self) -> int: """ Gibt die Anzahl verbleibender Slots im Fenster zurück. Returns: Anzahl verfügbarer Slots. """ with self._lock: self._cleanup(time.monotonic()) return max(0, self._max_requests - len(self._timestamps)) def time_until_available(self) -> float: """ Gibt die Zeit bis zum nächsten freien Slot zurück. Returns: Wartezeit in Sekunden (0 wenn sofort verfügbar). """ now = time.monotonic() with self._lock: self._cleanup(now) if len(self._timestamps) < self._max_requests: return 0.0 if self._timestamps: oldest = self._timestamps[0] return max(0, (oldest + self._window_size) - now) return 0.0 def reset(self) -> None: """Leert das Fenster.""" with self._lock: self._timestamps.clear() def __repr__(self) -> str: """String-Repräsentation.""" return ( f"SlidingWindow(window_size={self._window_size}, " f"max_requests={self._max_requests}, " f"current={self.current_count})" ) class SlidingWindowLimiter: """ Rate-Limiter basierend auf Sliding-Windows. Verwaltet mehrere Fenster für verschiedene Schlüssel. """ def __init__( self, window_size: float, max_requests: int, key_ttl: float = 3600.0, ) -> None: """ Initialisiert den Limiter. Args: window_size: Fenstergröße in Sekunden. max_requests: Maximum pro Fenster pro Schlüssel. key_ttl: Zeit bis ungenutzte Fenster gelöscht werden. """ self._window_size = window_size self._max_requests = max_requests self._key_ttl = key_ttl self._windows: dict[str, tuple[SlidingWindow, float]] = {} self._lock = threading.Lock() def _get_window(self, key: str) -> SlidingWindow: """Holt oder erstellt ein Fenster für einen Schlüssel.""" now = time.monotonic() with self._lock: if key in self._windows: window, _ = self._windows[key] self._windows[key] = (window, now) return window window = SlidingWindow( window_size=self._window_size, max_requests=self._max_requests, ) self._windows[key] = (window, now) return window def is_allowed(self, key: str) -> bool: """ Prüft, ob ein Request erlaubt ist. Args: key: Identifikator für den Rate-Limit-Kontext. Returns: True wenn erlaubt. """ window = self._get_window(key) return window.is_allowed() def get_remaining(self, key: str) -> int: """ Gibt die verbleibenden Requests für einen Schlüssel zurück. Args: key: Identifikator. Returns: Anzahl verbleibender Requests. """ window = self._get_window(key) return window.remaining() def get_retry_after(self, key: str) -> float: """ Gibt die Wartezeit bis zum nächsten freien Slot zurück. Args: key: Identifikator. Returns: Wartezeit in Sekunden. """ window = self._get_window(key) return window.time_until_available() def wait_and_proceed( self, key: str, max_wait: float | None = None, ) -> bool: """ Wartet und reserviert dann einen Slot. Args: key: Identifikator. max_wait: Maximale Wartezeit. Returns: True wenn erfolgreich. """ window = self._get_window(key) return window.wait_and_proceed(max_wait) def cleanup(self) -> int: """ Entfernt abgelaufene Fenster. Returns: Anzahl der entfernten Fenster. """ now = time.monotonic() expired_keys = [] with self._lock: for key, (_, last_access) in self._windows.items(): if now - last_access > self._key_ttl: expired_keys.append(key) for key in expired_keys: del self._windows[key] return len(expired_keys) def reset(self, key: str | None = None) -> None: """ Setzt ein oder alle Fenster zurück. Args: key: Optionaler spezifischer Schlüssel. """ with self._lock: if key: if key in self._windows: window, last_access = self._windows[key] window.reset() else: for window, _ in self._windows.values(): window.reset() def get_stats(self, key: str) -> SlidingWindowStats | None: """ Gibt Statistiken für einen Schlüssel zurück. Args: key: Identifikator. Returns: Statistiken oder None. """ with self._lock: if key in self._windows: window, _ = self._windows[key] return window.stats return None @property def window_count(self) -> int: """Gibt die Anzahl aktiver Fenster zurück.""" with self._lock: return len(self._windows) class SlidingWindowLog(SlidingWindow): """ Erweiterte Sliding-Window-Implementierung mit Logging. Speichert zusätzlich Metadaten zu jedem Request für Debugging und Analyse. """ def __init__( self, window_size: float, max_requests: int, log_size: int = 1000, ) -> None: """ Initialisiert das Sliding-Window mit Logging. Args: window_size: Fenstergröße in Sekunden. max_requests: Maximale Requests pro Fenster. log_size: Maximale Anzahl gespeicherter Log-Einträge. """ super().__init__(window_size, max_requests) self._log: deque[dict[str, Any]] = deque(maxlen=log_size) def is_allowed(self, metadata: dict[str, Any] | None = None) -> bool: """ Prüft, ob ein Request erlaubt ist und loggt ihn. Args: metadata: Optionale Metadaten zum Request. Returns: True wenn erlaubt. """ result = super().is_allowed() entry = { "timestamp": datetime.now().isoformat(), "allowed": result, "current_count": self.current_count, "remaining": self.remaining(), } if metadata: entry["metadata"] = metadata self._log.append(entry) return result def get_log( self, limit: int | None = None, allowed_only: bool = False, denied_only: bool = False, ) -> list[dict[str, Any]]: """ Gibt das Log zurück. Args: limit: Optionale Begrenzung der Einträge. allowed_only: Nur erlaubte Requests. denied_only: Nur abgelehnte Requests. Returns: Liste von Log-Einträgen. """ entries = list(self._log) if allowed_only: entries = [e for e in entries if e["allowed"]] elif denied_only: entries = [e for e in entries if not e["allowed"]] if limit: entries = entries[-limit:] return entries def clear_log(self) -> None: """Löscht das Log.""" self._log.clear()