from dataclasses import dataclass, field from typing import List, Optional, Type, Union import pybreaker from redis.asyncio import ConnectionPool, Redis, RedisCluster from redis.asyncio.multidb.database import Database, Databases from redis.asyncio.multidb.failover import ( DEFAULT_FAILOVER_ATTEMPTS, DEFAULT_FAILOVER_DELAY, AsyncFailoverStrategy, WeightBasedFailoverStrategy, ) from redis.asyncio.multidb.failure_detector import ( AsyncFailureDetector, FailureDetectorAsyncWrapper, ) from redis.asyncio.multidb.healthcheck import ( DEFAULT_HEALTH_CHECK_DELAY, DEFAULT_HEALTH_CHECK_INTERVAL, DEFAULT_HEALTH_CHECK_POLICY, DEFAULT_HEALTH_CHECK_PROBES, HealthCheck, HealthCheckPolicies, PingHealthCheck, ) from redis.asyncio.retry import Retry from redis.backoff import ExponentialWithJitterBackoff, NoBackoff from redis.data_structure import WeightedList from redis.event import EventDispatcher, EventDispatcherInterface from redis.multidb.circuit import ( DEFAULT_GRACE_PERIOD, CircuitBreaker, PBCircuitBreakerAdapter, ) from redis.multidb.failure_detector import ( DEFAULT_FAILURE_RATE_THRESHOLD, DEFAULT_FAILURES_DETECTION_WINDOW, DEFAULT_MIN_NUM_FAILURES, CommandFailureDetector, ) DEFAULT_AUTO_FALLBACK_INTERVAL = 120 def default_event_dispatcher() -> EventDispatcherInterface: return EventDispatcher() @dataclass class DatabaseConfig: """ Dataclass representing the configuration for a database connection. This class is used to store configuration settings for a database connection, including client options, connection sourcing details, circuit breaker settings, and cluster-specific properties. It provides a structure for defining these attributes and allows for the creation of customized configurations for various database setups. Attributes: weight (float): Weight of the database to define the active one. client_kwargs (dict): Additional parameters for the database client connection. from_url (Optional[str]): Redis URL way of connecting to the database. from_pool (Optional[ConnectionPool]): A pre-configured connection pool to use. circuit (Optional[CircuitBreaker]): Custom circuit breaker implementation. grace_period (float): Grace period after which we need to check if the circuit could be closed again. health_check_url (Optional[str]): URL for health checks. Cluster FQDN is typically used on public Redis Enterprise endpoints. Methods: default_circuit_breaker: Generates and returns a default CircuitBreaker instance adapted for use. """ weight: float = 1.0 client_kwargs: dict = field(default_factory=dict) from_url: Optional[str] = None from_pool: Optional[ConnectionPool] = None circuit: Optional[CircuitBreaker] = None grace_period: float = DEFAULT_GRACE_PERIOD health_check_url: Optional[str] = None def default_circuit_breaker(self) -> CircuitBreaker: circuit_breaker = pybreaker.CircuitBreaker(reset_timeout=self.grace_period) return PBCircuitBreakerAdapter(circuit_breaker) @dataclass class MultiDbConfig: """ Configuration class for managing multiple database connections in a resilient and fail-safe manner. Attributes: databases_config: A list of database configurations. client_class: The client class used to manage database connections. command_retry: Retry strategy for executing database commands. failure_detectors: Optional list of additional failure detectors for monitoring database failures. min_num_failures: Minimal count of failures required for failover failure_rate_threshold: Percentage of failures required for failover failures_detection_window: Time interval for tracking database failures. health_checks: Optional list of additional health checks performed on databases. health_check_interval: Time interval for executing health checks. health_check_probes: Number of attempts to evaluate the health of a database. health_check_delay: Delay between health check attempts. failover_strategy: Optional strategy for handling database failover scenarios. failover_attempts: Number of retries allowed for failover operations. failover_delay: Delay between failover attempts. auto_fallback_interval: Time interval to trigger automatic fallback. event_dispatcher: Interface for dispatching events related to database operations. Methods: databases: Retrieves a collection of database clients managed by weighted configurations. Initializes database clients based on the provided configuration and removes redundant retry objects for lower-level clients to rely on global retry logic. default_failure_detectors: Returns the default list of failure detectors used to monitor database failures. default_health_checks: Returns the default list of health checks used to monitor database health with specific retry and backoff strategies. default_failover_strategy: Provides the default failover strategy used for handling failover scenarios with defined retry and backoff configurations. """ databases_config: List[DatabaseConfig] client_class: Type[Union[Redis, RedisCluster]] = Redis command_retry: Retry = Retry( backoff=ExponentialWithJitterBackoff(base=1, cap=10), retries=3 ) failure_detectors: Optional[List[AsyncFailureDetector]] = None min_num_failures: int = DEFAULT_MIN_NUM_FAILURES failure_rate_threshold: float = DEFAULT_FAILURE_RATE_THRESHOLD failures_detection_window: float = DEFAULT_FAILURES_DETECTION_WINDOW health_checks: Optional[List[HealthCheck]] = None health_check_interval: float = DEFAULT_HEALTH_CHECK_INTERVAL health_check_probes: int = DEFAULT_HEALTH_CHECK_PROBES health_check_delay: float = DEFAULT_HEALTH_CHECK_DELAY health_check_policy: HealthCheckPolicies = DEFAULT_HEALTH_CHECK_POLICY failover_strategy: Optional[AsyncFailoverStrategy] = None failover_attempts: int = DEFAULT_FAILOVER_ATTEMPTS failover_delay: float = DEFAULT_FAILOVER_DELAY auto_fallback_interval: float = DEFAULT_AUTO_FALLBACK_INTERVAL event_dispatcher: EventDispatcherInterface = field( default_factory=default_event_dispatcher ) def databases(self) -> Databases: databases = WeightedList() for database_config in self.databases_config: # The retry object is not used in the lower level clients, so we can safely remove it. # We rely on command_retry in terms of global retries. database_config.client_kwargs.update( {"retry": Retry(retries=0, backoff=NoBackoff())} ) if database_config.from_url: client = self.client_class.from_url( database_config.from_url, **database_config.client_kwargs ) elif database_config.from_pool: database_config.from_pool.set_retry( Retry(retries=0, backoff=NoBackoff()) ) client = self.client_class.from_pool( connection_pool=database_config.from_pool ) else: client = self.client_class(**database_config.client_kwargs) circuit = ( database_config.default_circuit_breaker() if database_config.circuit is None else database_config.circuit ) databases.add( Database( client=client, circuit=circuit, weight=database_config.weight, health_check_url=database_config.health_check_url, ), database_config.weight, ) return databases def default_failure_detectors(self) -> List[AsyncFailureDetector]: return [ FailureDetectorAsyncWrapper( CommandFailureDetector( min_num_failures=self.min_num_failures, failure_rate_threshold=self.failure_rate_threshold, failure_detection_window=self.failures_detection_window, ) ), ] def default_health_checks(self) -> List[HealthCheck]: return [ PingHealthCheck(), ] def default_failover_strategy(self) -> AsyncFailoverStrategy: return WeightBasedFailoverStrategy()