o iU@svddlZddlZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZddlmZddlmZmZdd lmZdd lmZdd lmZdd lmZdd lmZmZmZm Z ddl!m"Z"m#Z#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3e4e5Z6e3GdddeeZ7defddZ8GdddeeZ9GdddZ:dS)N) as_completed)ThreadPoolExecutor)AnyCallableListOptional)BackgroundScheduler) NoBackoff)PubSubWorkerThread) CoreCommandsRedisModuleCommands)MaintNotificationsConfig)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheck MultiDbConfig)Database Databases SyncDatabase)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector) HealthCheckHealthCheckPolicy)GeoFailoverReason)Retry) experimentalc@s(eZdZdZdefddZddZdefdd Zd e dd fd d Z d8de de fddZ de de fddZd efddZd e defddZdefddZdefdd Zd!d"Zd#d$Zd%ed&gd ffd'd(Zd)d*Zd e de fd+d,Zdeee ffd-d.Zd/d0Zd1e d2e!d3e!fd4d5Z"d6d7Z#d S)9 MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc Cs||_|js |n|j|_|j|_|j|j |j |_ |j s%| n|j |_|jdur2|n|j|_|j|j|j|_|j|_|j|_|jtft|j|j|j|j|j|j|j|jd|_d|_t !|_"t#|_$||_%dS)N)failure_detectors databases command_retryfailover_strategyfailover_attemptsfailover_delayevent_dispatcherauto_fallback_intervalF)&r$ _databases health_checksdefault_health_checks_health_checkshealth_check_interval_health_check_intervalhealth_check_policyvaluehealth_check_probeshealth_check_probes_delay_health_check_policyr#default_failure_detectors_failure_detectorsr&default_failover_strategy_failover_strategy set_databasesr*_auto_fallback_intervalr)_event_dispatcherr%_command_retryupdate_supported_errorsConnectionRefusedErrorrr'r(command_executor initialized threadingRLock_hc_lockr _bg_scheduler_config)selfr"rHH/home/ubuntu/.local/lib/python3.10/site-packages/redis/multidb/client.py__init__+sH      zMultiDBClient.__init__cCsp||j|j|jd}|jD]\}}|j|j|jj t j kr,|s,||j _ d}q|s3tdd|_dS)zT Perform initialization of databases to define their initial state. FTz4Initial connection failed - no active database foundN)_perform_initial_health_checkrE run_recurringr0_check_databases_healthr+circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr@_active_databaserrA)rGis_active_db_founddatabaseweightrHrHrI initializeUs" zMultiDBClient.initializereturncCs|jS)zE Returns a sorted (by weight) list of all databases. )r+rGrHrHrI get_databaseswszMultiDBClient.get_databasesrVNcCsvd}|jD] \}}||krd}nq|std|||jjtjkr7|jdd\}}|tj f|j _ dSt d)zL Promote one of the existing databases to become an active. NT/Given database is not a member of database listrz1Cannot set active database, database is unhealthy) r+ ValueError_check_db_healthrNrQrRrS get_top_nrMANUALr@active_databaser)rGrVexists existing_db_highest_weighted_dbrHrHrIset_active_database}s$ z!MultiDBClient.set_active_databaseTskip_initial_health_checkcCstdtd|jd<d|jvrtdd|jd<|jr(|jjj|jfi|j}n"|jr@|jtdtd|jjj|jd}n |jjd i|j}|j durS| n|j }t |||j |j d }z||Wn tys|sqYnw|jd d\}}|j||j |||dS) z Adds a new database to the database list. Args: config: DatabaseConfig object that contains the database configuration. skip_initial_health_check: If True, adds the database even if it is unhealthy. r)retriesbackoffretrymaint_notifications_configF)enabled)connection_poolN)clientrNrWhealth_check_urlr]rH)rr client_kwargsr from_urlrF client_class from_pool set_retryrNdefault_circuit_breakerrrWrpr_rr+r`add_change_active_database)rGr"rhrorNrVrfhighest_weightrHrHrI add_databasesH   zMultiDBClient.add_database new_databasehighest_weight_databasecCs4|j|jkr|jjtjkr|tjf|j_dSdSdSN) rWrNrQrRrSr AUTOMATICr@rb)rGr{r|rHrHrIrxs  z%MultiDBClient._change_active_databasecCsP|j|}|jdd\}}||kr$|jjtjkr&|tjf|j _ dSdSdS)z< Removes a database from the database list. r]rN) r+remover`rNrQrRrSrrar@rb)rGrVrWrfryrHrHrIremove_databases  zMultiDBClient.remove_databaserWcCshd}|jD] \}}||krd}nq|std|jdd\}}|j||||_|||dS)z< Updates a database from the database list. NTr\r]r)r+r^r` update_weightrWrx)rGrVrWrcrdrerfryrHrHrIupdate_database_weightsz$MultiDBClient.update_database_weightfailure_detectorcCs|j|dS)z> Adds a new failure detector to the database. N)r7append)rGrrHrHrIadd_failure_detectorsz"MultiDBClient.add_failure_detector healthcheckcCs8|j|j|WddS1swYdS)z: Adds a new health check to the database. N)rDr.r)rGrrHrHrIadd_health_checks"zMultiDBClient.add_health_checkcOs |js||jj|i|S)zB Executes a single command and return its result. )rArXr@execute_commandrGargsoptionsrHrHrIr szMultiDBClient.execute_commandcCst|S)z: Enters into pipeline mode of the client. )PipelinerZrHrHrIpipelineszMultiDBClient.pipelinefuncrcOs&|js||jj|g||RS)z3 Executes callable as transaction. )rArXr@execute_transaction)rGrwatchesrrHrHrI transactionszMultiDBClient.transactioncKs|js|t|fi|S)z Return a Publish/Subscribe object. With this object, you can subscribe to channels and listen for messages that get published to them. )rArXPubSub)rGkwargsrHrHrIpubsub%szMultiDBClient.pubsubcCsP|j|j|}|s|jjtjkrtj|j_|S|r&|jjtjkr&tj|j_|S)zO Runs health checks on the given database until first failure. )r5executer.rNrQrROPENrS)rGrV is_healthyrHrHrIr_0s  zMultiDBClient._check_db_healthc sttjd`fddjD}i}z>t|jdD]4}z ||}|||<WqtyR}z|j}tj |j _ t j d|jdd||<WYd}~qd}~wwWn ty_td wWd|S1skwY|S) zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workerscs i|] \}}j||qSrH)submitr_).0rVreexecutorrGrHrI Gsz9MultiDBClient._check_databases_health..)timeoutz%Health check failed, due to exception)exc_infoFNz4Health check execution exceeds health_check_interval)rlenr+rr0resultrrVrRrrNrQloggerdebugoriginal_exception TimeoutError)rGfuturesresultsfuturerVe unhealthy_dbrHrrIrM@sB     z%MultiDBClient._check_databases_healthcCs|}d}|jjtjkrd|v}n!|jjtjkr(t|t|dk}n |jjtj kr5d|v}|s@t d|jjdS)zj Runs initial health check and evaluate healthiness based on initial_health_check_policy. TFz:Initial health check failed. Initial health check policy: N) rMrFinitial_health_check_policyr ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumr ONE_AVAILABLEr)rGrrrHrHrIrKes  z+MultiDBClient._perform_initial_health_checkrN old_state new_statecCs|tjkr ||jdS|tjkr)|tjkr)td|jd|j t t ||tjkr?|tjkrAt d|jddSdSdS)Nz Database z- is unreachable. Failover has been initiated.z is reachable again.) rR HALF_OPENr_rVrSrrwarningrErun_oncer_half_open_circuitinfo)rGrNrrrHrHrIrP}s   z/MultiDBClient._on_circuit_state_change_callbackcCs.|jr|j|jjr|jjjdSdS)z: Closes the client and all its resources. N)rEstopr@rbrocloserZrHrHrIrs  zMultiDBClient.close)T)$__name__ __module__ __qualname____doc__rrJrXrr[rrgrboolrzrxrrfloatrrrrrrrrrrr_dictrMrKrrRrPrrHrHrHrIr!$sH*" 8     % r!rNcCs tj|_dSr})rRrrQ)rNrHrHrIr rc@seZdZdZdefddZdddZdd Zd d Zde fd d Z de fddZ dddZ dddZdddZddZdeefddZdS)rzG Pipeline implementation for multiple logical Redis databases. rocCsg|_||_dSr})_command_stack_client)rGrorHrHrIrJs zPipeline.__init__rYcC|Sr}rHrZrHrHrI __enter__zPipeline.__enter__cC |dSr}reset)rGexc_type exc_value tracebackrHrHrI__exit__rzPipeline.__exit__cC$z|WdStyYdSwr}r ExceptionrZrHrHrI__del__s  zPipeline.__del__cCs t|jSr})rrrZrHrHrI__len__ zPipeline.__len__cCsdS)z1Pipeline instances should always evaluate to TrueTrHrZrHrHrI__bool__szPipeline.__bool__NcCs g|_dSr})rrZrHrHrIrrzPipeline.resetcCs |dS)zClose the pipelineNrrZrHrHrIr zPipeline.closecOs|j||f|S)ar Stage a command to be executed when execute() is next called Returns the current Pipeline object back so commands can be chained together, such as: pipe = pipe.set('foo', 'bar').incr('baz').decr('bang') At some other point, you can then run: pipe.execute(), which will execute all commands queued in the pipe. )rrrrHrHrIpipeline_execute_commands z!Pipeline.pipeline_execute_commandcOs|j|i|S)zAdds a command to the stack)rrGrrrHrHrIrszPipeline.execute_commandcCs<|jjs |jz|jjt|jW|S|w)z0Execute all the commands in the current pipeline)rrArXr@execute_pipelinetuplerrrZrHrHrIrs zPipeline.execute)rYrrYN)rrrrr!rJrrrintrrrrrrrrrrrHrHrHrIrs    rc @seZdZdZdefddZd.ddZd/d d Zd/d d Zd/d dZ e de fddZ ddZ ddZddZddZddZddZddZ d0d!e d"efd#d$Z d0d!e d"efd%d&Z   d1d'ed(e d)eed*e dd+f d,d-ZdS)2rz2 PubSub object for multi database client. rocKs||_|jjjdi|dS)zInitialize the PubSub object for a multi-database client. Args: client: MultiDBClient instance to use for pub/sub operations **kwargs: Additional keyword arguments to pass to the underlying pubsub implementation NrH)rr@r)rGrorrHrHrIrJszPubSub.__init__rYcCrr}rHrZrHrHrIrrzPubSub.__enter__NcCrr}rrZrHrHrIrs  zPubSub.__del__cCs|jjdS)Nrrr@execute_pubsub_methodrZrHrHrIrsz PubSub.resetcCrr}rrZrHrHrIrrz PubSub.closecCs |jjjjSr})rr@ active_pubsub subscribedrZrHrHrIrrzPubSub.subscribedcG|jjjdg|RS)NrrrGrrHrHrIrs zPubSub.execute_commandcO|jjjdg|Ri|S)aE Subscribe to channel patterns. Patterns supplied as keyword arguments expect a pattern name as the key and a callable as the value. A pattern's callable will be invoked automatically when a message is received on that pattern rather than producing a message via ``listen()``. psubscriberrrHrHrIr zPubSub.psubscribecGr)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrrHrHrIr zPubSub.punsubscribecOr)aR Subscribe to channels. Channels supplied as keyword arguments expect a channel name as the key and a callable as the value. A channel's callable will be invoked automatically when a message is received on that channel rather than producing a message via ``listen()`` or ``get_message()``. subscriberrrHrHrIr!rzPubSub.subscribecGr)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrrHrHrIr-szPubSub.unsubscribecOr)az Subscribes the client to the specified shard channels. Channels supplied as keyword arguments expect a channel name as the key and a callable as the value. A channel's callable will be invoked automatically when a message is received on that channel rather than producing a message via ``listen()`` or ``get_sharded_message()``. ssubscriberrrHrHrIr4rzPubSub.ssubscribecGr)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrrHrHrIr@rzPubSub.sunsubscribeFignore_subscribe_messagesrcC|jjjd||dS)a Get the next message if one is available, otherwise None. If timeout is specified, the system will wait for `timeout` seconds before returning. Timeout should be specified as a floating point number, or None, to wait indefinitely. get_messagerrrrGrrrHrHrIrI  zPubSub.get_messagecCr)a& Get the next message if one is available in a sharded channel, otherwise None. If timeout is specified, the system will wait for `timeout` seconds before returning. Timeout should be specified as a floating point number, or None, to wait indefinitely. get_sharded_messagerrrrHrHrIrYrzPubSub.get_sharded_message sleep_timedaemonexception_handlersharded_pubsubr cCs|jjj|||||dS)N)rrrr)rr@execute_pubsub_run)rGrrrrrHrHrI run_in_threadiszPubSub.run_in_thread)rYrr)Fr)rFNF)rrrrr!rJrrrrpropertyrrrrrrrrrrrrrrrrHrHrHrIrsV         r);loggingrBconcurrent.futuresrconcurrent.futures.threadrtypingrrrrredis.backgroundr redis.backoffr redis.clientr redis.commandsr r redis.maint_notificationsr redis.multidb.circuitrrrRredis.multidb.command_executorrredis.multidb.configrrrrredis.multidb.databaserrrredis.multidb.exceptionrrrredis.multidb.failure_detectorrredis.multidb.healthcheckrrredis.observability.attributesr redis.retryr redis.utilsr getLoggerrrr!rrrrHrHrHrIs:              wC