125 Zeilen
3.4 KiB
Python
125 Zeilen
3.4 KiB
Python
"""
|
|
Rate Limit Service Interface - Domain Service für Rate Limiting
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime, timedelta
|
|
|
|
from domain.value_objects.action_timing import ActionTiming, ActionType
|
|
from domain.entities.rate_limit_policy import RateLimitPolicy
|
|
|
|
|
|
class IRateLimitService(ABC):
|
|
"""
|
|
Interface für Rate Limit Service.
|
|
Definiert die Geschäftslogik für adaptives Rate Limiting.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def calculate_delay(self, action_type: ActionType, context: Optional[Dict[str, Any]] = None) -> float:
|
|
"""
|
|
Berechnet die optimale Verzögerung für eine Aktion.
|
|
|
|
Args:
|
|
action_type: Typ der auszuführenden Aktion
|
|
context: Optionaler Kontext (z.B. Platform, Session-ID)
|
|
|
|
Returns:
|
|
Verzögerung in Sekunden
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def record_action(self, timing: ActionTiming) -> None:
|
|
"""
|
|
Zeichnet eine ausgeführte Aktion für Analyse auf.
|
|
|
|
Args:
|
|
timing: Timing-Informationen der Aktion
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def detect_rate_limit(self, response: Any) -> bool:
|
|
"""
|
|
Erkennt ob eine Response auf Rate Limiting hindeutet.
|
|
|
|
Args:
|
|
response: HTTP Response oder Browser-Seite
|
|
|
|
Returns:
|
|
True wenn Rate Limit erkannt wurde
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_policy(self, action_type: ActionType) -> RateLimitPolicy:
|
|
"""
|
|
Holt die aktuelle Rate Limit Policy für einen Action Type.
|
|
|
|
Args:
|
|
action_type: Typ der Aktion
|
|
|
|
Returns:
|
|
Rate Limit Policy
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def update_policy(self, action_type: ActionType, policy: RateLimitPolicy) -> None:
|
|
"""
|
|
Aktualisiert die Rate Limit Policy für einen Action Type.
|
|
|
|
Args:
|
|
action_type: Typ der Aktion
|
|
policy: Neue Policy
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_statistics(self,
|
|
action_type: Optional[ActionType] = None,
|
|
timeframe: Optional[timedelta] = None) -> Dict[str, Any]:
|
|
"""
|
|
Holt Statistiken über Rate Limiting.
|
|
|
|
Args:
|
|
action_type: Optional - nur für spezifischen Action Type
|
|
timeframe: Optional - nur für bestimmten Zeitraum
|
|
|
|
Returns:
|
|
Dictionary mit Statistiken
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def reset_statistics(self) -> None:
|
|
"""Setzt alle gesammelten Statistiken zurück."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_action_allowed(self, action_type: ActionType) -> bool:
|
|
"""
|
|
Prüft ob eine Aktion basierend auf Rate Limits erlaubt ist.
|
|
|
|
Args:
|
|
action_type: Typ der Aktion
|
|
|
|
Returns:
|
|
True wenn Aktion erlaubt ist
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def wait_if_needed(self, action_type: ActionType) -> float:
|
|
"""
|
|
Wartet die notwendige Zeit bevor eine Aktion ausgeführt werden kann.
|
|
|
|
Args:
|
|
action_type: Typ der Aktion
|
|
|
|
Returns:
|
|
Tatsächlich gewartete Zeit in Sekunden
|
|
"""
|
|
pass |