/ 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/ddl0m1Z1ddl2m3Z3ej4e5Z6e3GddeeZ7defdZ8GddeeZ9GddZ: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) experimentalceZdZdZdefdZdZdefdZde ddfd Z d%de d e fd Z d e de fdZdefdZde defdZdefdZdefdZdZdZdedgdffdZdZde de fdZdeee ffdZdZd e d!e!d"e!fd#Z"d$Z#dS)& MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc N||_|js|n|j|_|j|_|j|j |j |_ |j s| n|j |_|j|n|j|_|j|j|j|_|j|_|j|_|jt4ft7|j|j|j|j|j|j|j|j|_d|_tAj!|_"tG|_$||_%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$s C:\Users\Dell Inspiron 16\Desktop\tws\AgrotaPowerBi\back-agrota-powerbi\mcp-client-agrota\venv\Lib\site-packages\redis/multidb/client.py__init__zMultiDBClient.__init__+s **,,' &F ( ( * * *%  '-&B#7=7Q7W7W  &(H8 8 ! + *F , , . . .) '/  , , . . .)  --do>>>'-'D$!'!8$2 335K4MNNN 6"5o-"5$6!0!3#'#? ! ! ! !!)) 022 cX||j|j|jd}|jD]N\}}|j|j|jj tj kr|s||j _ d}O|stdd|_dS)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 U U U%77@R7:B%6%)"! *F  rMreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r.rJs rK get_databaseszMultiDBClient.get_databasesws rMrZNcLd}|jD]\}}||krd}n|std|||jjt jkr=|jdd\}}|tj 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. ValueError_check_db_healthrRrUrVrW get_top_nrMANUALrCactive_databaser)rJrZexists existing_db_highest_weighted_dbs rKset_active_databasez!MultiDBClient.set_active_database}s"o  NKh&&' PNOO O h'''   !W^ 3 3%)_%>%>q%A%A!%D " !(5D ! 1 F& ?   rMTskip_initial_health_checkc\tdt|jd<d|jvrtd|jd<|jr#|jjj|jfi|j}ny|jr[|jtdt|jj|j}n|jjd i|j}|j | n|j }t|||j |j } ||n#t$r|sYnwxYw|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)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(M(M(MW% (v/C C C(777  != > ? G7T\.7#)#7FF  G   & &uQ 'L'L'L M M M\.88 & 09FF/T\.FF1EFFF~%  * * , , ,  =#4       ! !( + + + +)   ,    /3o.G.G.J.J1.M+^ Hho666 $$X/BCCCCCs*E EE new_databasehighest_weight_databasec|j|jkr4|jjtjkr|t jf|j_dSdSdSN) r[rRrUrVrWr AUTOMATICrCrh)rJrrs rKrz%MultiDBClient._change_active_databasesV  "9"@ @ @$*gn<<!+5D ! 1 1 1 A @< Adds a new failure detector to the database. N)r:append)rJrs rKadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rM healthcheckcz|j5|j|ddddS#1swxYwYdS)z: Adds a new health check to the database. N)rGr1r)rJrs rKadd_health_checkzMultiDBClient.add_health_checks] 4 4   & &{ 3 3 3 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4s 044c\|js||jj|i|S)zB Executes a single command and return its result. )rDr\rCexecute_commandrJargsoptionss rKrzMultiDBClient.execute_command s:  OO   4t$4dFgFFFrMc t|S)z: Enters into pipeline mode of the client. )Pipeliner_s rKpipelinezMultiDBClient.pipelines~~rMfuncrcb|js||jj|g||RS)z3 Executes callable as transaction. )rDr\rCexecute_transaction)rJrwatchesrs rK transactionzMultiDBClient.transactionsB  OO   8t$8RR'RRRRrMc R|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. )rDr\PubSub)rJkwargss rKpubsubzMultiDBClient.pubsub%s5   OO   d%%f%%%rMc|j|j|}|s2|jjt jkrt j|j_|S|r0|jjt jkrt j|j_|S)zO Runs health checks on the given database until first failure. )r8executer1rRrUrVOPENrW)rJrZ is_healthys rKrezMultiDBClient._check_db_health0sz .66t7JHUU  4%55)0 &   4H,2gnDD%,^H  "rMcttj5fdjD}i} t|jD]|} ||}|||<##t $rM}|j}tj |j _ t d|jd||<Yd}~ud}~wwxYwn#t$rtdwxYw dddn #1swxYwY|S) zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workerscNi|]!\}}j||"Srx)submitre).0rZrkexecutorrJs rK z9MultiDBClient._check_databases_health..Gs@Ha 5x@@(rM)timeoutz%Health check failed, due to exception)exc_infoFNz4Health check execution exceeds health_check_interval)rlenr.rr3resultrrZrVrrRrUloggerdebugoriginal_exception TimeoutError)rJfuturesresultsfuturerZe unhealthy_dbrs` @rKrQz%MultiDBClient._check_databases_health@s  C,@,@ A A A X#'?G G *T%@66F 6#*6?,2MMOO))5 6 6 6'(z 5<\ ,2 C%&%9% 16 ------ 6 6     "J !6               <sSC;CA65C6 C ACCC  CC;C++C;;C?C?c|}d}|jjtjkrd|v}n|jjtjkr6t|t|dz k}n0|jjtj krd|v}|std|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) rQrIinitial_health_check_policyr ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumr ONE_AVAILABLEr)rJrrs rKrOz+MultiDBClient._perform_initial_health_checkes..00 < 37I7W W Wgnn&6&66JJ L 4!4 5 5W^^--..W1AAJJ L 48J8X X X!1!11J /wT\Muww   rMrR old_state new_statec|tjkr||jdS|tjkrY|tjkrIt d|jd|j tt||tjkr5|tjkr't d|jddSdSdS)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 3 3 3 F  & &9 +D+D NN[G,[[[      ' '$&8'     & &9+F+F KKJG$4JJJ K K K K K ' &+F+FrMc|jr|j|jjr%|jjjdSdS)z: Closes the client and all its resources. N)rHstoprCrhrvcloser_s rKrzMultiDBClient.closes`   &   # # % % %  0 A  ! 1 8 > > @ @ @ @ @ A ArM)T)$__name__ __module__ __qualname____doc__rrLr\rr`rrmrboolrrrrfloatrrrrrrrrrrredictrQrOrrVrTrrxrMrKr#r#$sU (}((((T    Dy  L T    :IM6D6D$6DAE6D6D6D6Dp ( CO     D|DUDDDD&9_9999 4K4444GGG S*t); <SSSS & & &$ #hn)=####J0L%L29LFMLLLL&AAAAArMr#rRc(tj|_dSr)rVrrU)rRs rKrrs%GMMMrMceZdZdZdefdZddZdZdZde fdZ de fd Z dd Z dd Zdd ZdZdeefdZd S)rzG Pipeline implementation for multiple logical Redis databases. rvc"g|_||_dSr)_command_stack_client)rJrvs rKrLzPipeline.__init__s  rMr]c|Srrxr_s rK __enter__zPipeline.__enter__ rMc.|dSrreset)rJexc_type exc_value tracebacks rK__exit__zPipeline.__exit__ rMcR |dS#t$rYdSwxYwrr Exceptionr_s rK__del__zPipeline.__del__s:  JJLLLLL    DD   &&c*t|jSr)rrr_s rK__len__zPipeline.__len__s4&'''rMcdS)z1Pipeline instances should always evaluate to TrueTrxr_s rK__bool__zPipeline.__bool__strMNcg|_dSr)rr_s rKrzPipeline.resets rMc.|dS)zClose the pipelineNrr_s rKrzPipeline.closes rMc>|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. )rrrs rKpipeline_execute_commandz!Pipeline.pipeline_execute_commands$ ""D'?333 rMc|j|i|S)zAdds a command to the stack)rrJrrs rKrzPipeline.execute_commands,t,d=f===rMc |jjs|j |jjt |j|S#|wxYw)z0Execute all the commands in the current pipeline)rrDr\rCexecute_pipelinetuplerrr_s rKrzPipeline.executesq|' & L # # % % % <0AAd)** JJLLLLDJJLLLLs 0A,,B)r]rr]N)rrrrr#rLrrrintrrrrrrrrrrrxrMrKrrs } ((((($!!!!    >>> c      rMrc eZdZdZdefdZddZddZddZdd Z e de fd Z d Z d Zd ZdZdZdZdZ d de defdZ d de defdZ d!dede deede ddf dZdS)"rz2 PubSub object for multi database client. rvc B||_|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 Nrx)rrCr)rJrvrs rKrLzPubSub.__init__s/ , %,66v66666rMr]c|Srrxr_s rKrzPubSub.__enter__rrMNcR |dS#t$rYdSwxYwrrr_s rKrzPubSub.__del__s<  JJLLLLL    DD rc@|jjdS)NrrrCexecute_pubsub_methodr_s rKrz PubSub.resets|,BB7KKKrMc.|dSrrr_s rKrz PubSub.closerrMc.|jjjjSr)rrC active_pubsub subscribedr_s rKrzPubSub.subscribeds|,:EErMc2|jjjdg|RS)NrrrJrs rKrzPubSub.execute_commands.Bt|,B  $    rMc8|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()``. psubscriberrs rKr zPubSub.psubscribe >Ct|,B     #)   rMc2|jjjdg|RS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rKr zPubSub.punsubscribe1 Ct|,B  !    rMc8|jjjdg|Ri|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 rKrzPubSub.subscribe!s>Ct|,B     "(   rMc2|jjjdg|RS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rKrzPubSub.unsubscribe-s& Ct|,B=XSWXXXXrMc8|jjjdg|Ri|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.ssubscribe4r rMc2|jjjdg|RS)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rKrzPubSub.sunsubscribe@rrMFignore_subscribe_messagesrcF|jjd||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   rMcF|jjd||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 cJ|jj|||||S)N)r r!rr")rrCexecute_pubsub_run)rJrr r!r"s rK run_in_threadzPubSub.run_in_threadis6|,?? /) @   rM)r]rr)Fr)rFNF)rrrrr#rLrrrrpropertyrrrr r rrrrrrrrrr%rxrMrKrrs 7} 7 7 7 7LLLLFDFFFXF             YYY       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#rrrrxrMrKr;s++++++888888000000000000000000######++++++<<<<<<<<>>>>>>000000222222AAAAAA EDDDDDDDDD ;:::::DDDDDDDD<<<<<<$$$$$$  8 $ $rArArArArA'rArArAj &&&&&@@@@@"L@@@FU U U U U U U U U U rM