"""Rate-Limiting-Modul für Rippy API.""" import time import secrets from collections import defaultdict from typing import Dict, Optional # Default Rate Limit MAX_REQUESTS_PER_MINUTE = 100 # In-Memory Rate Limit Store (in Produktion mit Redis) rate_limit_store: Dict[str, list] = defaultdict(list) def check_rate_limit(client_id: str, max_requests: int = MAX_REQUESTS_PER_MINUTE, window_seconds: int = 60) -> bool: """Prüfe ob Client rate-limited ist.""" current_time = time.time() window_start = current_time - window_seconds # Bereinige alte Einträge rate_limit_store[client_id] = [ timestamp for timestamp in rate_limit_store[client_id] if timestamp > window_start ] # Prüfe ob Limit erreicht if len(rate_limit_store[client_id]) >= max_requests: return False # Füge neuen Request hinzu rate_limit_store[client_id].append(current_time) return True def get_rate_limit_remaining(client_id: str, max_requests: int = MAX_REQUESTS_PER_MINUTE) -> int: """Hole verbleibende Requests.""" current_time = time.time() window_start = current_time - 60 current_count = len([ timestamp for timestamp in rate_limit_store[client_id] if timestamp > window_start ]) return max(0, max_requests - current_count) def reset_rate_limit(client_id: str) -> None: """Setze Rate Limit für Client zurück.""" rate_limit_store[client_id] = [] # API-Key Store (in Produktion mit Datenbank) api_keys: Dict[str, Dict] = { "example_key": { "key": "example_key", "name": "Beispiel API Key", "created_at": time.time(), "rate_limit": 100 } } def validate_api_key(api_key: str) -> Optional[Dict]: """Validiere API Key.""" if api_key in api_keys: return api_keys[api_key] return None def create_api_key(name: str) -> Dict: """Erstelle neuer API Key.""" key = secrets.token_urlsafe(32) api_keys[key] = { "key": key, "name": name, "created_at": time.time(), "rate_limit": MAX_REQUESTS_PER_MINUTE } return api_keys[key] def delete_api_key(key: str) -> bool: """Lösche API Key.""" if key in api_keys: del api_keys[key] return True return False