h^GddlZddlZddlmZddlmZddlmZmZm 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 ||_||_|j|j|j|j|_|j |j |j |_ | |_|j|j|j|j|n|j|_|j|j|j|_|j|_|j|_|jt6ft9|j|j|j|j|j|j|j|j|_d|_ tCj"|_#tI|_%||_&dS)N)failure_detectors databases command_retryfailover_strategyfailover_attemptsfailover_delayevent_dispatcherauto_fallback_intervalF)'r _databasesdefault_health_checks_health_checks health_checksextendhealth_check_interval_health_check_intervalhealth_check_policyvaluehealth_check_probeshealth_check_probes_delay_health_check_policydefault_failure_detectors_failure_detectorsrr"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 T/var/www/html/auto_sub_bot/venv/lib/python3.11/site-packages/redis/multidb/client.py__init__zMultiDBClient.__init__s **,,$::<<   +   & &v'; < < <&,&B#7=7Q7W7W  &(H8 8 !#)"B"B"D"D  # /  # * *6+C D D D'/  , , . . .)  --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 rEraise_exception_on_failed_hcz>MultiDBClient.initialize..raise_exception_on_failed_hcLsKrG)on_errorFTz4Initial connection failed - no active database foundN)_check_databases_healthrB run_recurringr-r'circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr=active_databaserr>)rDrMis_active_db_founddatabaseweights rE initializezMultiDBClient.initializeGs     $$.J$KKK ((  '  (   # $ * * Hf   - -d.T U U U%77@R78@%5%)"! *F  rGreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r'rDs rE get_databaseszMultiDBClient.get_databasesjs rGrYNc4d}|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_healthrQrTrUrV get_top_nr=rWr)rDrYexists existing_db_highest_weighted_dbs rEset_active_databasez!MultiDBClient.set_active_databaseps"o  NKh&&' PNOO O h'''   !W^ 3 3%)_%>%>q%A%A!%D " 4 Adds a new failure detector to the database. N)r4append)rDrys rEadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rG healthcheckcz|j5|j|ddddS#1swxYwYdS)z: Adds a new health check to the database. N)rAr)r{)rDr}s rEadd_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>r[r=execute_commandrDargsoptionss rErzMultiDBClient.execute_commands:  OO   4t$4dFgFFFrGc t|S)z: Enters into pipeline mode of the client. )Pipeliner^s rEpipelinezMultiDBClient.pipelines~~rGfuncrcb|js||jj|g||RS)z3 Executes callable as transaction. )r>r[r=execute_transaction)rDrwatchesrs rE transactionzMultiDBClient.transactionsB  OO   8t$8RR'RRRRrGc 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>r[PubSub)rDkwargss rEpubsubzMultiDBClient.pubsubs5   OO   d%%f%%%rGc|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. )r2executer)rQrTrUOPENrV)rDrY is_healthys rErdzMultiDBClient._check_db_healthsz .66t7JHUU  4%55)0 &   4H,2gnDD%,^H  "rGrNcttj5fdjD} 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|!SrK)submitrd).0rYrhexecutorrDs rE z8MultiDBClient._check_databases_health..s>Ha 5x@@rG)timeoutz%Health check failed, due to exception)exc_infoNz4Health check execution exceeds health_check_interval)rlenr'rr-resultrrYrUrrQrTlogger exceptionoriginal_exception TimeoutError)rDrNfuturesfuturee unhealthy_dbrs` @rErOz%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<<DDrQ old_state new_statec|tjkr||jdS|tjkr8|tjkr*|jtt|dSdSdSrJ) rU HALF_OPENrdrYrVrrBrun_oncer_half_open_circuit)rDrQrrs rErSz/MultiDBClient._on_circuit_state_change_callback"s ) ) )  ! !'"2 3 3 3 F  & &9 +D+D   ' '$&8'      ' &+D+DrGcL|jjjdSrJ)r=rWclientcloser^s rErzMultiDBClient.close.s# -4::<<<<|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. )rr{rs rEpipeline_execute_commandz!Pipeline.pipeline_execute_commandYs$ ""D'?333 rGc|j|i|S)zAdds a command to the stack)rrDrrs rErzPipeline.execute_commandhs,t,d=f===rGc |jjs|j |jjt |j|S#|wxYw)z0Execute all the commands in the current pipeline)rr>r[r=execute_pipelinetuplerrr^s rErzPipeline.executelsq|' & L # # % % % <0AAd)** JJLLLLDJJLLLLs 0A,,B)r\rr\N)rrrrrrFrrrintrrrrrrrrrrrKrGrErr6s } ((((($!!!!    >>> c      rGrc 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 NrK)rr=r)rDrrs rErFzPubSub.__init__~s/ , %,66v66666rGr\c|SrJrKr^s rErzPubSub.__enter__rrGNcR |dS#t$rYdSwxYwrJrr^s rErzPubSub.__del__s<  JJLLLLL    DD rc@|jjdS)Nrrr=execute_pubsub_methodr^s rErz PubSub.resets|,BB7KKKrGc.|dSrJrr^s rErz PubSub.closerrGc.|jjjjSrJ)rr= active_pubsub subscribedr^s rErzPubSub.subscribeds|,:EErGc2|jjjdg|RS)NrrrDrs rErzPubSub.execute_commands.Bt|,B  $    rGc8|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 rErzPubSub.psubscribe>Ct|,B     #)   rGc2|jjjdg|RS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rErzPubSub.punsubscribe1 Ct|,B  !    rGc8|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 rErzPubSub.subscribes>Ct|,B     "(   rGc2|jjjdg|RS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rErzPubSub.unsubscribes& Ct|,B=XSWXXXXrGc8|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 rErzPubSub.ssubscriberrGc2|jjjdg|RS)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rErzPubSub.sunsubscriberrGFignore_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_messagerrrrDrrs rErzPubSub.get_messages0|,BB &?C   rGcF|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 rErzPubSub.get_sharded_messages0|,BB !&?C   rG sleep_timedaemonexception_handlersharded_pubsubr cJ|jj|||||S)N)rrrr)rr=execute_pubsub_run)rDrrrrs rE run_in_threadzPubSub.run_in_threads6|,?? /) @   rG)r\rr)Fr)rFNF)rrrrrrFrrrrpropertyrrrrrrrrrrrrrrrrKrGrErrys 7} 7 7 7 7LLLLFDFFFXF             YYY       IL  )- @E    "IL  )- @E    $ 04$       $H-           rGr)0loggingr?concurrent.futuresrconcurrent.futures.threadrtypingrrrrredis.backgroundr redis.clientr redis.commandsr r redis.multidb.circuitr rrUredis.multidb.command_executorrredis.multidb.configrrredis.multidb.databaserrrredis.multidb.exceptionrrredis.multidb.failure_detectorrredis.multidb.healthcheckrr redis.utilsr getLoggerrrrrrrrKrGrErs?++++++888888000000000000000000++++++<<<<<<<<000000222222AAAAAADDDDDDDDDDDDDDDDDDXXXXXXXX::::::DDDDDDDD$$$$$$  8 $ $W=W=W=W=W='W=W=W=t&&&&&@@@@@"L@@@FU U U U U U U U U U rG