o run_coro_sync_perform_initial_health_checkrun_recurring_coror._check_databases_healthr)circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr<_active_databaserr=)rCis_active_db_founddatabaseweightrDrDrE initialize]s"  zMultiDBClient.initializereturncCs|jS)zE Returns a sorted (by weight) list of all databases. )r)rKrDrDrE get_databasesszMultiDBClient.get_databasesrYNcCs|d}|jD] \}}||krd}nq|std|j|j||jjtjkr:|j dd\}}|t j f|j _ dStd)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) ValueErrorr>rM_check_db_healthrQrTrUrV get_top_nrMANUALr<active_databaser)rCrYexists existing_db_highest_weighted_dbrDrDrEset_active_databases$z!MultiDBClient.set_active_databaseTskip_initial_health_checkcCs"tdtd|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 |j|j|Wn tyv|stYnw|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)clientrQrZhealth_check_urlr_rD)rr client_kwargsr from_urlrB client_class from_pool set_retryrQdefault_circuit_breakerrrZrrr>rMrarr)rbadd_change_active_database)rCr rjrqrQrYrhhighest_weightrDrDrE add_databasesH   zMultiDBClient.add_database new_databasehighest_weight_databasecCs4|j|jkr|jjtjkr|tjf|j_dSdSdSrH) rZrQrTrUrVr AUTOMATICr<rd)rCr}r~rDrDrErzs  z%MultiDBClient._change_active_databasecCsP|j|}|jdd\}}||kr$|jjtjkr&|tjf|j _ dSdSdS)z< Removes a database from the database list. r_rN) r)removerbrQrTrUrVrrcr<rd)rCrYrZrhr{rDrDrEremove_databases  zMultiDBClient.remove_databaserZcCshd}|jD] \}}||krd}nq|std|jdd\}}|j||||_|||dS)z< Updates a database from the database list. NTr^r_r)r)r`rb update_weightrZrz)rCrYrZrerfrgrhr{rDrDrEupdate_database_weightsz$MultiDBClient.update_database_weightfailure_detectorcCs|j|dS)z> Adds a new failure detector to the database. N)r3append)rCrrDrDrEadd_failure_detector sz"MultiDBClient.add_failure_detector healthcheckcCs8|j|j|WddS1swYdS)z: Adds a new health check to the database. N)rAr,r)rCrrDrDrEadd_health_checks"zMultiDBClient.add_health_checkcOs |js||jj|i|S)zB Executes a single command and return its result. )r=r[r<execute_commandrCargsoptionsrDrDrErszMultiDBClient.execute_commandcCst|S)z: Enters into pipeline mode of the client. )PipelinerKrDrDrEpipeline"szMultiDBClient.pipelinefuncrcOs&|js||jj|g||RS)z3 Executes callable as transaction. )r=r[r<execute_transaction)rCrwatchesrrDrDrE transaction(szMultiDBClient.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. )r=r[PubSub)rCkwargsrDrDrEpubsub1szMultiDBClient.pubsubcs|j t|j}Wdn1swY|j||IdH}|s3|jjtjkr1tj|j_|S|rA|jjtj krAtj |j_|S)zO Runs health checks on the given database until first failure. N) rAlistr,r1executerQrTrUOPENrV)rCrYr* is_healthyrDrDrEra<s   zMultiDBClient._check_db_healthcsig|_|jD]\}}t||}||<|j|q tj|jddiIdH}fddt|j|D}|D]\}}t |t rZ|j }t j |j_tjd|jdd||<q=|S) zk Runs health checks as a recurring task. Runs health checks against all databases. return_exceptionsTNcsi|] \}}||qSrDrD).0taskresult task_to_dbrDrE _sz9MultiDBClient._check_databases_health..z%Health check failed, due to exception)exc_infoF) _hc_tasksr)asyncio create_taskrargatherzipitems isinstancerrYrUrrQrTloggerdebugoriginal_exception)rCrYrgrresults db_resultsr unhealthy_dbrDrrErPOs,    z%MultiDBClient._check_databases_healthcs|IdH}d}|jjtjkrd|v}n!|jjtjkr,t|t|dk}n |jjtj kr9d|v}|sDt d|jjdS)zj Runs initial health check and evaluate healthiness based on initial_health_check_policy. NTFz:Initial health check failed. Initial health check policy: ) rPrBinitial_health_check_policyr ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumlen ONE_AVAILABLEr)rCrrrDrDrErNqs   z+MultiDBClient._perform_initial_health_checkrQ old_state new_statecCs|tjkr|j|j|jdS|tjkr,|tjkr,t d|jd|j t t ||tjkrB|tjkrDt d|jddSdSdS)Nz Database z- is unreachable. Failover has been initiated.z is reachable again.)rU HALF_OPENr>run_coro_fire_and_forgetrarYrVrrwarningrun_oncer_half_open_circuitinfo)rCrQrrrDrDrErSs  z/MultiDBClient._on_circuit_state_change_callbackcCsV|jrz |j|jjWn tyYnw|j|jjr)|jjjdSdS)z: Closes the client and all its resources. N) r>rMr1rIrJstopr<rdrqrKrDrDrErIs  zMultiDBClient.close)T)%__name__ __module__ __qualname____doc__rrFrLr[rr]rrirboolr|rzrrfloatrrrrrrrrrrradictrPrNrrUrSrIrDrDrDrEr#sJ* & 8     " rrQcCs tj|_dSrH)rUrrT)rQrDrDrEr rc@seZdZdZdefddZdddZdd Zd d Zde fd d Z de fddZ dddZ dddZdddZddZdeefddZdS)rzG Pipeline implementation for multiple logical Redis databases. rqcCsg|_||_dSrH)_command_stack_client)rCrqrDrDrErFs zPipeline.__init__r\cC|SrHrDrKrDrDrE __enter__zPipeline.__enter__cC |dSrHreset)rCexc_type exc_value tracebackrDrDrE__exit__rzPipeline.__exit__cCrGrHrrJrKrDrDrErLs  zPipeline.__del__cCs t|jSrH)rrrKrDrDrE__len__ zPipeline.__len__cCsdS)z1Pipeline instances should always evaluate to TrueTrDrKrDrDrE__bool__szPipeline.__bool__NcCs g|_dSrH)rrKrDrDrErrzPipeline.resetcCs |dS)zClose the pipelineNrrKrDrDrErI 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. )rrrrDrDrEpipeline_execute_commands z!Pipeline.pipeline_execute_commandcOs|j|i|S)zAdds a command to the stack)rrCrrrDrDrErszPipeline.execute_commandcCs<|jjs |jz|jjt|jW|S|w)z0Execute all the commands in the current pipeline)rr=r[r<execute_pipelinetuplerrrKrDrDrErs zPipeline.execute)r\rr\N)rrrrrrFrrrLintrrrrrIrrrrrrDrDrDrErs    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. rqcKs||_|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 NrD)rr<r)rCrqrrDrDrErFszPubSub.__init__r\cCrrHrDrKrDrDrErrzPubSub.__enter__NcCrGrHrrKrDrDrErL s  zPubSub.__del__cCs|jjdS)Nrrr<execute_pubsub_methodrKrDrDrErsz PubSub.resetcCrrHrrKrDrDrErIrz PubSub.closecCs |jjjjSrH)rr< active_pubsub subscribedrKrDrDrErrzPubSub.subscribedcG|jjjdg|RS)NrrrCrrDrDrErs 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()``. psubscriberrrDrDrEr#zPubSub.psubscribecGr)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrrDrDrEr/ 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()``. subscriberrrDrDrEr8rzPubSub.subscribecGr)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrrDrDrErDszPubSub.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()``. ssubscriberrrDrDrErKrzPubSub.ssubscribecGr)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrrDrDrErWrzPubSub.sunsubscribeFignore_subscribe_messagestimeoutcC|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_messagerrrrCrrrDrDrEr`  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_messagerrrrDrDrErprzPubSub.get_sharded_message sleep_timedaemonexception_handlersharded_pubsubr cCs|jjj|||||dS)N)rrrr)rr<execute_pubsub_run)rCrrrrrDrDrE run_in_threadszPubSub.run_in_thread)r\rr)Fr)rFNF)rrrrrrFrrLrrIpropertyrrrrrrrrrrrrrrrrDrDrDrErsV         r)8rloggingr?typingrrrr!redis.asyncio.multidb.healthcheckrrredis.backgroundr redis.backoffr redis.clientr redis.commandsr r redis.maint_notificationsr redis.multidb.circuitrrrUredis.multidb.command_executorrredis.multidb.configrrrrredis.multidb.databaserrrredis.multidb.exceptionrrrredis.multidb.failure_detectorrredis.observability.attributesr redis.retryr redis.utilsr getLoggerrrrrrrrDrDrDrEs:            C