OǻiUddlZddlZddlmZddlmZddlmZmZm 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/ddl0m1Z1ddl2m3Z3ejhe5Z6e3GddeeZ7defdZ8GddeeZ9GddZ:y)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) experimentalceZdZdZdefdZdZdefdZde ddfd Z d$de d e fd Z d e d e fdZdefdZde defdZdefdZdefdZdZdZdedgdffdZdZde de fdZdeee ffdZdZde d e!d!e!fd"Z"d#Z#y)% MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc |j|_|js|jn |j|_|j |_|jj|j|j|_ |js|jn |j|_|j|j!n |j|_|j"j%|j|j&|_|j*|_|j.|_|j0j3t4ft7|j|j|j0|j"|j8|j:|j,|j(|_d|_tAjB|_"tG|_$||_%y)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$s R/opt/lhia/marcimex/agent/venv/lib/python3.12/site-packages/redis/multidb/client.py__init__zMultiDBClient.__init__+s **,''  ( ( *%%  '-&B&B#7=7Q7Q7W7W  & &(H(H8 ! ++  , , .)) ''/  , , .))  --doo>'-'D'D$!'!8!8$22 335K4MN 6"55oo--"55$66!00!33#'#?#? ! !!) 02 c|j|jj|j|jd}|j D]h\}}|j j|j|j jtjk(sS|rV||j_ d}j|s tdd|_y)zT Perform initialization of databases to define their initial state. FTz4Initial connection failed - no active database foundN)_perform_initial_health_checkrH run_recurringr3_check_databases_healthr.circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDrC_active_databaserrD)rJis_active_db_founddatabaseweights rK initializezMultiDBClient.initializeUs **, ((  ' '  ( ( # $ * Hf    - -d.T.T U%%7@R:B%%6%)" *"*F  rMreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r.rJs rK get_databaseszMultiDBClient.get_databaseswsrMrZNcfd}|jD]\}}||k(s d}n|s td|j||jjt j k(rC|jjdd\}}|tjf|j_ ytd)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_healthrRrUrVrW get_top_nrMANUALrCactive_databaser)rJrZexists existing_db_highest_weighted_dbs rKset_active_databasez!MultiDBClient.set_active_database}s"oo NKh&  NO O h'    ! !W^^ 3%)__%>%>q%A!%D " !((5D ! ! 1 & ?  rMskip_initial_health_checkctdt|jd<d|jvrtd|jd<|jr<|j j j|jfi|j}n|jr_|jjtdt|j j j|j}n&|j j d i|j}|j|jn |j}t|||j|j } |j||j j#d d\}}|j j%||j|j'||y#t$r|sYhwxYw) 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)clientrRr[health_check_urlrc)r r client_kwargsrfrom_urlrI client_class from_pool set_retryrRdefault_circuit_breakerrr[rwrerr.rfadd_change_active_database)rJr$rnrvrRrZrlhighest_weights rK add_databasezMultiDBClient.add_databases).a(MW% (v/C/C C(7  != > ??7T\\..77#)#7#7F      & &uQ 'L M\\..88 & 0 09F/T\\..F1E1EFF~~%  * * ,  ==#44     ! !( + /3oo.G.G.J1.M+^ Hhoo6 $$X/BC * ,- s/G G)(G) new_databasehighest_weight_databasec|j|jkDrJ|jjtjk(r"|t j f|j_yyyN) r[rRrUrVrWr AUTOMATICrCrh)rJrrs rKrz%MultiDBClient._change_active_databasesZ   "9"@"@ @$$**gnn<!++5D ! ! 1= ArMc|jj|}|jjdd\}}||krJ|jjt j k(r"|tjf|j_ yyy)z< Removes a database from the database list. rcrN) r.removerfrRrUrVrWrrgrCrh)rJrZr[rlrs rKremove_databasezMultiDBClient.remove_databases''1.2oo.G.G.J1.M+^ f $#++11W^^C$!((5D ! ! 1D %rMr[cd}|jD]\}}||k(s d}n|s td|jjdd\}}|jj||||_|j ||y)z< Updates a database from the database list. NTrbrcr)r.rdrf update_weightr[r)rJrZr[rirjrkrlrs rKupdate_database_weightz$MultiDBClient.update_database_weights"oo NKh&  NO O.2oo.G.G.J1.M+^ %%h7  $$X/BCrMfailure_detectorc:|jj|y)z> Adds a new failure detector to the database. N)r:append)rJrs rKadd_failure_detectorz"MultiDBClient.add_failure_detectors &&'78rM healthcheckc||j5|jj|dddy#1swYyxYw)z: Adds a new health check to the database. N)rGr1r)rJrs rKadd_health_checkzMultiDBClient.add_health_checks4]] 4    & &{ 3 4 4 4s2;cr|js|j|jj|i|S)zB Executes a single command and return its result. )rDr\rCexecute_commandrJargsoptionss rKrzMultiDBClient.execute_command s5 OO 4t$$44dFgFFrMct|S)z: Enters into pipeline mode of the client. )Pipeliner_s rKpipelinezMultiDBClient.pipelines~rMfuncrcx|js|j|jj|g||S)z3 Executes callable as transaction. )rDr\rCexecute_transaction)rJrwatchesrs rK transactionzMultiDBClient.transactions: OO 8t$$88RR'RRrMc R|js|jt|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. )rDr\PubSub)rJkwargss rKpubsubzMultiDBClient.pubsub%s'  OO d%f%%rMcv|jj|j|}|sH|jjt j k7rt j |j_|S|rF|jjt jk7rt j|j_|S)zO Runs health checks on the given database until first failure. )r8executer1rRrUrVOPENrW)rJrZ is_healthys rKrezMultiDBClient._check_db_health0s ..66t7J7JHU %%5)0  &  H,,22gnnD%,^^H   "rMc ^tt|j5}|jDcic]"\}}|j|j||$}}}i} t ||j D]} ||}|j||< ddd|Scc}}w#t$r[}|j}tj|j_ tjd|j d||<Yd}~d}~wwxYw#t"$r t#dwxYw#1swYSxYw)zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workers)timeoutz%Health check failed, due to exception)exc_infoFNz4Health check execution exceeds health_check_interval)rlenr.submitrerr3resultrrZrVrrRrUloggerdebugoriginal_exception TimeoutError) rJexecutorrZrkfuturesresultsfuturee unhealthy_dbs rKrQz%MultiDBClient._check_databases_health@s6  C,@ A X$(??Ha 5 5x@(JG G *T%@%@6F 6#*6?,2MMO) 6 <96 6'(zz 5<\\ ,,2 C%&%9%9% 16 - 6  "J 5 <sYD"'BD"D 8B#D D"# D,AD=D DD  DD""D,c|j}d}|jjtjk(rd|j v}n|jjtj k(r)t|j t|dz kD}n9|jjtjk(rd|j v}|s"td|jjy)zj Runs initial health check and evaluate healthiness based on initial_health_check_policy. TFz:Initial health check failed. Initial health check policy: N) rQrIinitial_health_check_policyr ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumr ONE_AVAILABLEr)rJrrs rKrOz+MultiDBClient._perform_initial_health_checkes..0 << 3 37I7W7W Wgnn&66J LL 4 4!44 5W^^-.W1AAJ LL 4 48J8X8X X!11J/LT\\MuMuLvw rMrR old_state new_statec|tjk(r|j|jy|tjk(r[|tj k(rHt jd|jd|jjtt||tjk7r8|tjk(r$t jd|jdyyy)Nz Database z- is unreachable. Failover has been initiated.z is reachable again.) rV HALF_OPENrerZrWrrwarningrHrun_oncer_half_open_circuitinfo)rJrRrrs rKrTz/MultiDBClient._on_circuit_state_change_callback}s )) )  ! !'"2"2 3   &9 +D NNG,,--Z[     ' '$&8'   &9+F KK)G$4$4#55IJ K,G &rMc|jr|jj|jjr/|jjjj yy)z: Closes the client and all its resources. N)rHstoprCrhrvcloser_s rKrzMultiDBClient.closesQ       # # %  0 0  ! ! 1 1 8 8 > > @ 1rM)T)$__name__ __module__ __qualname____doc__rrLr\rr`rrmrboolrrrrfloatrrrrrrrrrrredictrQrOrrVrTrrxrMrKr#r#$s' (}(T  Dy  L T :IM6D$6DAE6Dp ( CO  D|DUD&9_9 4K4G S*t); <S &$ #hn)=#J0L%L29LFML&ArMr#rRc.tj|_yr)rVrrU)rRs rKrrs%%GMrMcxeZdZdZdefdZddZdZdZde fdZ de fd Z dd Z dd Zdd ZdZdeefdZy )rzG Pipeline implementation for multiple logical Redis databases. rvc g|_||_yr)_command_stack_client)rJrvs rKrLzPipeline.__init__s  rMr]c|Srrxr_s rK __enter__zPipeline.__enter__ rMc$|jyrreset)rJexc_type exc_value tracebacks rK__exit__zPipeline.__exit__  rMcD |jy#t$rYywxYwrr Exceptionr_s rK__del__zPipeline.__del__s"  JJL     c,t|jSr)rrr_s rK__len__zPipeline.__len__s4&&''rMcy)z1Pipeline instances should always evaluate to TrueTrxr_s rK__bool__zPipeline.__bool__srMNcg|_yr)rr_s rKrzPipeline.resets  rMc$|jy)zClose the pipelineNrr_s rKrzPipeline.closes  rMc@|jj||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. )rrrs rKpipeline_execute_commandz!Pipeline.pipeline_execute_commands! ""D'?3 rMc&|j|i|S)zAdds a command to the stack)rrJrrs rKrzPipeline.execute_commands,t,,d=f==rMc|jjs|jj |jjj t |j |jS#|jwxYw)z0Execute all the commands in the current pipeline)rrDr\rCexecute_pipelinetuplerrr_s rKrzPipeline.executes_||'' LL # # % <<00AAd))* JJLDJJLs 7A::B )r]rr]N)rrrrr#rLrrrintrrrrrrrrrrrxrMrKrrs^} (($! > c rMrc eZdZdZdefdZddZddZddZdd Z e de fd Z d Z d Zd ZdZdZdZdZ dde defdZ dde defdZ ddede deede ddf dZy) rz2 PubSub object for multi database client. rvc ^||_|jjjdi|y)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 Nrx)rrCr)rJrvrs rKrLzPubSub.__init__s( , %%,,6v6rMr]c|Srrxr_s rKrzPubSub.__enter__rrMNcD |jy#t$rYywxYwrrr_s rKrzPubSub.__del__s$  JJL   rcL|jjjdS)NrrrCexecute_pubsub_methodr_s rKrz PubSub.resets||,,BB7KKrMc$|jyrrr_s rKrz PubSub.closerrMcV|jjjjSr)rrC active_pubsub subscribedr_s rKrzPubSub.subscribeds||,,::EEErMcP|jjjdg|S)NrrrJrs rKrzPubSub.execute_commands,Bt||,,BB  $  rMcV|jjjdg|i|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()``. psubscriberrs rKrzPubSub.psubscribe 7Ct||,,BB   #)  rMcP|jjjdg|S)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rKr zPubSub.punsubscribe/ Ct||,,BB  !  rMcV|jjjdg|i|S)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()``. subscriberrs rKr zPubSub.subscribe!s7Ct||,,BB   "(  rMcP|jjjdg|S)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rKrzPubSub.unsubscribe-s( Ct||,,BB=XSWXXrMcV|jjjdg|i|S)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()``. ssubscriberrs rKrzPubSub.ssubscribe4rrMcP|jjjdg|S)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rKrzPubSub.sunsubscribe@r rMignore_subscribe_messagesrcR|jjjd||S)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_messagerrrrJrrs rKrzPubSub.get_messageIs0||,,BB &?C  rMcR|jjjd||S)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_messagerrrs rKrzPubSub.get_sharded_messageYs0||,,BB !&?C  rM sleep_timedaemonexception_handlersharded_pubsubr cV|jjj|||||S)N)rrrr)rrCexecute_pubsub_run)rJrrrrs rK run_in_threadzPubSub.run_in_threadis6||,,?? /) @  rM)r]rr)F)r"FNF)rrrrr#rLrrrrpropertyrrrrr r rrrrrrrrr!rxrMrKrrs 7} 7LFDFF     Y   IL )- @E "IL )- @E $ 04$      $H-      rMr);loggingrEconcurrent.futuresrconcurrent.futures.threadrtypingrrrrredis.backgroundr redis.backoffr redis.clientr redis.commandsr r redis.maint_notificationsrredis.multidb.circuitrrrVredis.multidb.command_executorrredis.multidb.configrrrrredis.multidb.databaserrrredis.multidb.exceptionrrrredis.multidb.failure_detectorrredis.multidb.healthcheckrrredis.observability.attributesr redis.retryr redis.utilsr! getLoggerrrr#rrrrxrMrKr8s+8000#+<>02A ED ;D<$   8 $rA'rArAj &&@"L@FU U rM