keJiHddlZddlZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZddlmZmZddlmZdd lmZdd lmZdd lmZmZdd lmZmZmZdd lm Z m!Z!ddl"m#Z#ddl$m%Z%m&Z&ddl'm(Z(ej)e*Z+e(GddeeZ,defdZ-GddeeZ.GddZ/dS)N) as_completed)ThreadPoolExecutor)AnyCallableListOptional)BackgroundScheduler)PubSubWorkerThread) CoreCommandsRedisModuleCommands)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIOD MultiDbConfig)Database Databases SyncDatabase)NoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector) HealthCheckHealthCheckPolicy) experimentalc eZdZdZdefdZdZdefdZde ddfd Z de fd Z d e d e fd Z de fdZde defdZdefdZdefdZdZdZdedgdffdZdZde defdZd#deegdffdZdeded efd!Z d"Z!dS)$ MultiDBClientzx Client that operates on multiple logical Redis databases. Should be used in Active-Active 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_policyrdefault_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)selfrs Z/var/www/html/movieo_spanner_bot/venv/lib/python3.11/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 cbd}|||j|j|jd}|jD]N\}}|j|j|jjtj kr|s||j _ d}O|stdd|_dS)zT Perform initialization of databases to define their initial state. c|N)errors rDraise_exception_on_failed_hcz>MultiDBClient.initialize..raise_exception_on_failed_hcMsKrF)on_errorFTz4Initial connection failed - no active database foundN)_check_databases_healthrA run_recurringr,r'circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr<active_databaserr=)rCrLis_active_db_founddatabaseweights rD initializezMultiDBClient.initializeHs     $$.J$KKK ((  '  (   # $ * * Hf   - -d.T U U U%77@R78@%5%)"! *F  rFreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r'rCs rD get_databaseszMultiDBClient.get_databasesks rFrXNc4d}|jD]\}}||krd}n|std|||jjt jkr1|jdd\}}||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_healthrPrSrTrU get_top_nr<rVr)rCrXexists existing_db_highest_weighted_dbs rDset_active_databasez!MultiDBClient.set_active_databaseqs"o  NKh&&' PNOO O h'''   !W^ 3 3%)_%>%>q%A%A!%D " 4 Adds a new failure detector to the database. N)r3append)rCrxs rDadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rF healthcheckcz|j5|j|ddddS#1swxYwYdS)z: Adds a new health check to the database. N)r@r*rz)rCr|s rDadd_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. )r=rZr<execute_commandrCargsoptionss rDrzMultiDBClient.execute_commands:  OO   4t$4dFgFFFrFc t|S)z: Enters into pipeline mode of the client. )Pipeliner]s rDpipelinezMultiDBClient.pipelines~~rFfuncrcb|js||jj|g||RS)z3 Executes callable as transaction. )r=rZr<execute_transaction)rCrwatchesrs rD transactionzMultiDBClient.transactionsB  OO   8t$8RR'RRRRrFc 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. )r=rZPubSub)rCkwargss rDpubsubzMultiDBClient.pubsubs5   OO   d%%f%%%rFc|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. )r1executer*rPrSrTOPENrU)rCrX is_healthys rDrczMultiDBClient._check_db_healthsz .66t7JHUU  4%55)0 &   4H,2gnDD%,^H  "rFrMcttj5fdjD} t|jD]~} |#t $rZ}|j}tj |j _ t d|j|r||jYd}~wd}~wwxYwn#t$rtdwxYw ddddS#1swxYwYdS)zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workerscLh|] \}}j|!SrJ)submitrc).0rXrgexecutorrCs rD z8MultiDBClient._check_databases_health..s>Ha 5x@@rF)timeoutz%Health check failed, due to exception)exc_infoNz4Health check execution exceeds health_check_interval)rlenr'rr,resultrrXrTrrPrSlogger exceptionoriginal_exception TimeoutError)rCrMfuturesfuturee unhealthy_dbrs` @rDrNz%MultiDBClient._check_databases_healths  C,@,@ A A A X#'?G  *T%@;;F ; 5 ; ; ;'(z 5<\ ,2((C%&%9) $;$HQ%9::: ; ;     "J !;                  sSC<CA)(C) C 3ACCC  CC<C++C<<DDrP old_state new_statec|tjkr||jdS|tjkr8|tjkr*|jtt|dSdSdSrI) rT HALF_OPENrcrXrUrrArun_oncer_half_open_circuit)rCrPrrs rDrRz/MultiDBClient._on_circuit_state_change_callback#s ) ) )  ! !'"2 3 3 3 F  & &9 +D+D   ' '$&8'      ' &+D+DrFc|jr|j|jjr%|jjjdSdS)z: Closes the client and all its resources. N)rAstopr<rVclientcloser]s rDrzMultiDBClient.close/s`   &   # # % % %  0 A  ! 1 8 > > @ @ @ @ @ A ArFrI)"__name__ __module__ __qualname____doc__rrErZrr^rrirnrlrrtfloatrwrr{rr~rrrrrboolrc ExceptionrNr rTrRrrJrFrDrrs: (}((((T! ! ! Fy  L T    2 D\ D D D DA(ACOAAAA H H H H HD|DUDDDD&9_9999 4K4444GGG S*t); <SSSS & & &$  )d9J0K    D % 29 FM    AAAAArFrrPc(tj|_dSrI)rTrrS)rPs rDrr9s%GMMMrFceZdZdZdefdZddZdZdZde fdZ de fd Z dd Z dd Zdd ZdZdeefdZd S)rzG Pipeline implementation for multiple logical Redis databases. rc"g|_||_dSrI)_command_stack_client)rCrs rDrEzPipeline.__init__Bs  rFr[c|SrIrJr]s rD __enter__zPipeline.__enter__F rFc.|dSrIreset)rCexc_type exc_value tracebacks rD__exit__zPipeline.__exit__I rFcR |dS#t$rYdSwxYwrIrrr]s rD__del__zPipeline.__del__Ls:  JJLLLLL    DD   &&c*t|jSrI)rrr]s rD__len__zPipeline.__len__Rs4&'''rFcdS)z1Pipeline instances should always evaluate to TrueTrJr]s rD__bool__zPipeline.__bool__UstrFNcg|_dSrI)rr]s rDrzPipeline.resetYs rFc.|dS)zClose the pipelineNrr]s rDrzPipeline.close\s rFc>|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. )rrzrs rDpipeline_execute_commandz!Pipeline.pipeline_execute_command`s$ ""D'?333 rFc|j|i|S)zAdds a command to the stack)rrCrrs rDrzPipeline.execute_commandos,t,d=f===rFc |jjs|j |jjt |j|S#|wxYw)z0Execute all the commands in the current pipeline)rr=rZr<execute_pipelinetuplerrr]s rDrzPipeline.executessq|' & L # # % % % <0AAd)** JJLLLLDJJLLLLs 0A,,B)r[rr[N)rrrrrrErrrintrrrrrrrrrrrJrFrDrr=s } ((((($!!!!    >>> c      rFrc 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. rc 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 NrJ)rr<r)rCrrs rDrEzPubSub.__init__s/ , %,66v66666rFr[c|SrIrJr]s rDrzPubSub.__enter__rrFNcR |dS#t$rYdSwxYwrIrr]s rDrzPubSub.__del__s<  JJLLLLL    DD rc@|jjdS)Nrrr<execute_pubsub_methodr]s rDrz PubSub.resets|,BB7KKKrFc.|dSrIrr]s rDrz PubSub.closerrFc.|jjjjSrI)rr< active_pubsub subscribedr]s rDrzPubSub.subscribeds|,:EErFc2|jjjdg|RS)NrrrCrs rDrzPubSub.execute_commands.Bt|,B  $    rFc8|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 rDrzPubSub.psubscribe>Ct|,B     #)   rFc2|jjjdg|RS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rDrzPubSub.punsubscribe1 Ct|,B  !    rFc8|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 rDrzPubSub.subscribes>Ct|,B     "(   rFc2|jjjdg|RS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rDrzPubSub.unsubscribes& Ct|,B=XSWXXXXrFc8|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 rDrzPubSub.ssubscriberrFc2|jjjdg|RS)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rDrzPubSub.sunsubscriberrFFignore_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_messagerrrrCrrs rDrzPubSub.get_messages0|,BB &?C   rFcF|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 rDrzPubSub.get_sharded_messages0|,BB !&?C   rF sleep_timedaemonexception_handlersharded_pubsubr cJ|jj|||||S)N)rrrr)rr<execute_pubsub_run)rCrrrrs rD run_in_threadzPubSub.run_in_threads6|,?? /) @   rF)r[rr)Fr)rFNF)rrrrrrErrrrpropertyrrrrrrrrrrrrrrrrJrFrDrrs 7} 7 7 7 7LLLLFDFFFXF             YYY       IL  )- @E    "IL  )- @E    $ 04$       $H-           rFr)0loggingr>concurrent.futuresrconcurrent.futures.threadrtypingrrrrredis.backgroundr redis.clientr redis.commandsr r redis.multidb.circuitr rrTredis.multidb.command_executorrredis.multidb.configrrredis.multidb.databaserrrredis.multidb.exceptionrrredis.multidb.failure_detectorrredis.multidb.healthcheckrr redis.utilsr getLoggerrrrrrrrJrFrDrsG++++++888888000000000000000000++++++<<<<<<<<000000222222AAAAAADDDDDDDDDDDDDDDDDDXXXXXXXX::::::DDDDDDDD$$$$$$  8 $ $^A^A^A^A^A'^A^A^AB &&&&&@@@@@"L@@@FU U U U U U U U U U rF