75 Zeilen
2.4 KiB
Python
75 Zeilen
2.4 KiB
Python
"""
|
|
Rate limit repository interface.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Optional, List, Dict
|
|
from datetime import datetime
|
|
|
|
from domain.entities.rate_limit_policy import RateLimitPolicy
|
|
|
|
|
|
class IRateLimitRepository(ABC):
|
|
"""Interface for rate limit data persistence."""
|
|
|
|
@abstractmethod
|
|
def save_policy(self, policy: RateLimitPolicy) -> None:
|
|
"""Save or update a rate limit policy."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_policy(self, platform: str, action: str) -> Optional[RateLimitPolicy]:
|
|
"""Get rate limit policy for platform and action."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_all_policies(self, platform: str = None) -> List[RateLimitPolicy]:
|
|
"""Get all policies, optionally filtered by platform."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def record_action(self, platform: str, action: str,
|
|
success: bool = True, proxy: str = None) -> None:
|
|
"""Record an action for rate limit tracking."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_action_count(self, platform: str, action: str,
|
|
window_minutes: int = 60,
|
|
proxy: str = None) -> int:
|
|
"""Get action count within time window."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_recent_actions(self, platform: str, action: str,
|
|
limit: int = 100) -> List[Dict[str, any]]:
|
|
"""Get recent actions for analysis."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_rate_limited(self, platform: str, action: str,
|
|
proxy: str = None) -> bool:
|
|
"""Check if action is currently rate limited."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_wait_time(self, platform: str, action: str,
|
|
proxy: str = None) -> int:
|
|
"""Get wait time in seconds before next action allowed."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def reset_limits(self, platform: str = None, action: str = None,
|
|
proxy: str = None) -> int:
|
|
"""Reset rate limits. Returns count of records affected."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_limit_status(self, platform: str) -> Dict[str, Dict[str, any]]:
|
|
"""Get current rate limit status for all actions on platform."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def cleanup_old_records(self, days_to_keep: int = 7) -> int:
|
|
"""Remove old rate limit records. Returns count deleted."""
|
|
pass |