keJiTHddlZddlZddlmZmZmZmZmZmZm Z ddl m Z ddl m Z ddlmZmZddlmZmZddlmZddlmZmZdd lmZdd lmZmZdd lmZdd lm Z!dd l"m#Z#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ej+e,Z-e*GddeeZ.defdZ/GddeeZ0GddZ1dS)N)Any AwaitableCallable CoroutineListOptionalUnion) PubSubHandler)DefaultCommandExecutor)DEFAULT_GRACE_PERIOD MultiDbConfig) AsyncDatabase Databases)AsyncFailureDetector) HealthCheckHealthCheckPolicy)BackgroundScheduler)AsyncCoreCommandsAsyncRedisModuleCommands)CircuitBreaker)State)NoValidDatabaseExceptionUnhealthyDatabaseException)ChannelT EncodableTKeyT) experimentalc eZdZdZdefdZd,dZdZdZde fd Z d e dd fd Z d e fd Z de de fdZd e fdZd e defdZdefdZdefdZdZdZd dd ddedgeeeeffdedeed ed!eef d"Z d#Z! d-d$eee"ge#eed fffd%Z$d e defd&Z%d'e&d(e'd)e'fd*Z(d+Z)d S). MultiDBClientzx Client that operates on multiple logical Redis databases. Should be used in Active-Active database setups. configc x||_|js|n|j|_|j|_|j|j |j |_ |j s| n|j |_|j|n|j|_|j|j|j|_|j|_|j|_|jt4gt7|j|j|j|j|j|j|j|j|_d|_tAj!|_"tG|_$||_%d|_&g|_'d|_(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_delay_health_check_policyr"default_failure_detectors_failure_detectorsr%default_failover_strategy_failover_strategy set_databasesr)_auto_fallback_intervalr(_event_dispatcherr$_command_retryupdate_supported_errorsConnectionRefusedErrorr r&r'command_executor initializedasyncioLock_hc_lockr _bg_scheduler_config_recurring_hc_task _hc_tasks_half_open_state_task)selfr s b/var/www/html/movieo_spanner_bot/venv/lib/python3.11/site-packages/redis/asyncio/multidb/client.py__init__zMultiDBClient.__init__s **,,' &F ( ( * * *%  '-&B#7=7Q7W7W  &(A8 8 ! + *F , , . . .) '/  , , . . .)  --do>>>'-'D$!'!8$2 335K4LMMM 6"5o-"5$6!0!3#'#? ! ! ! !  022 "&%)"""rIreturncLK|js|d{V|SN)r@ initializerIs rJ __aenter__zMultiDBClient.__aenter__Ks8 $//## # # # # # # # rLcK|jr|j|jr|j|jD]}|dSrO)rFcancelrHrG)rIexc_type exc_value tracebackhc_tasks rJ __aexit__zMultiDBClient.__aexit__Pst  " -  # * * , , ,  % 0  & - - / / /~  G NN      rLcKd}||d{Vtj|j|j|j|_d}|jD]b\}}|j |j |jj tj kr$|s"|j|d{Vd}c|st!dd|_dS)zT Perform initialization of databases to define their initial state. c K|rO)errors rJraise_exception_on_failed_hcz>MultiDBClient.initialize..raise_exception_on_failed_hc]s KrL)on_errorNFTz4Initial connection failed - no active database found)_check_databases_healthrA create_taskrDrun_recurring_asyncr/rFr*circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr?set_active_databaserr@)rIr^is_active_db_founddatabaseweights rJrPzMultiDBClient.initializeXs*    **4P*QQQQQQQQQ#*"5   2 2+,  # # # $ * * Hf   - -d.T U U U%77@R7+??IIIIIIIII%)"! *F  rLc|jS)zE Returns a sorted (by weight) list of all databases. )r*rQs rJ get_databaseszMultiDBClient.get_databases}s rLrkNclKd}|jD]\}}||krd}n|std||d{V|jjt jkrE|jdd\}}|j |d{VdStd)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_healthrcrfrgrh get_top_nr?rir)rIrkexists existing_db_highest_weighted_dbs rJriz!MultiDBClient.set_active_databases"o  NKh&&' PNOO O##H---------   !W^ 3 3%)_%>%>q%A%A!%D " ';;HEE E E E E E E E F& ?   rLcBK|jD]\}}||krtd||d{V|jdd\}}|j||j|||d{VdS)z; Adds a new database to the database list. zGiven database already existsNrqr)r*rrrsrtaddrl_change_active_database)rIrkrvrwrxhighest_weights rJ add_databasezMultiDBClient.add_databases#o B BNKh&& !@AAA'##H---------.2o.G.G.J.J1.M+^ Hho666**85HIIIIIIIIIIIrL new_databasehighest_weight_databasecK|j|jkr<|jjtjkr$|j|d{VdSdSdSrO)rlrcrfrgrhr?ri)rIr~rs rJr{z%MultiDBClient._change_active_databasesk  "9"@ @ @$*gn<<';;LII I I I I I I I I I A @< Adds a new failure detector to the database. N)r6append)rIrs rJadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rL healthcheckcK|j4d{V|j|dddd{VdS#1d{VswxYwYdS)z: Adds a new health check to the database. N)rCr-r)rIrs rJadd_health_checkzMultiDBClient.add_health_checks= 4 4 4 4 4 4 4 4   & &{ 3 3 3 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4 4s> A AcxK|js|d{V|jj|i|d{VS)zB Executes a single command and return its result. N)r@rPr?execute_commandrIargsoptionss rJrzMultiDBClient.execute_commandsb $//## # # # # # # #:T*:DLGLLLLLLLLLrLc t|S)z: Enters into pipeline mode of the client. )PipelinerQs rJpipelinezMultiDBClient.pipelines~~rLF shard_hintvalue_from_callable watch_delayfuncrwatchesrrrcK|js|d{V|jj|g|R|||dd{VS)z3 Executes callable as transaction. Nr)r@rPr?execute_transaction)rIrrrrrs rJ transactionzMultiDBClient.transactions $//## # # # # # # #>T*>    " 3#           rLcbK|js|d{Vt|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. N)r@rPPubSub)rIkwargss rJpubsubzMultiDBClient.pubsubsK  $//## # # # # # # #d%%f%%%rLr_cK fdjD_tjtjjddijd{V}n'#tj$rtjdwxYw|D]g}t|trP|j }tj |j _ td|j|r||jhdS) zk Runs health checks as a recurring task. Runs health checks against all databases. cdg|],\}}tj|-Sr\)rArars).0rkrwrIs rJ z9MultiDBClient._check_databases_health..sFHa#D$9$9($C$CDDrLreturn_exceptionsT)timeoutNz4Health check execution exceeds health_check_intervalz%Health check failed, due to exception)exc_info)r*rGrAwait_forgatherr/ TimeoutError isinstancerrkrgOPENrcrflogger exceptionoriginal_exception)rIr_resultsresult unhealthy_dbs` rJr`z%MultiDBClient._check_databases_healthsS #'?DN$,^&*3 GG#   &F    8 8F&"<== 8% -4\ $*  ;#6! 8HV6777 8 8s A A$A7c"K|j|j|d{V}|s2|jjt jkrt j|j_|S|r0|jjt jkrt j|j_|S)zO Runs health checks on the given database until first failure. N)r4executer-rcrfrgrrh)rIrk is_healthys rJrszMultiDBClient._check_db_health6s  4<<           4%55)0 &   4H,2gnDD%,^H  "rLrc old_state new_statec<tj}|tjkr3tj||j|_dS|tjkr3|tj kr%| tt|dSdSdSrO) rAget_running_looprg HALF_OPENrarsrkrHrhr call_laterr _half_open_circuit)rIrcrrloops rJrez/MultiDBClient._on_circuit_state_change_callbackHs')) ) ) ))0)<%%g&677**D & F  & &9 +D+D OO02Dg N N N N N ' &+D+DrLcxK|jjr+|jjjd{VdSdSrO)r?active_databaseclientacloserQs rJrzMultiDBClient.acloseVsT  0 H'7>EEGG G G G G G G G G G H HrL)rIrrMrrO)*__name__ __module__ __qualname____doc__r rKrRrYrPrrnrrir}r{rfloatrrrrrrrrr rrrrstrboolrr Exceptionrr`rsrrgrerr\rLrJrrs ,*},*,*,*,*\ # # # Jy  - D    2 J= J J J JJ)JDQJJJJ Qm Q Q Q QJ]JEJJJJ&95I9999 4+4444MMM%)$)'+     |U3 #+>%??@  SM  "  e_     , & & &PT$8$88YK3T>1J$JKL$8$8$8$8L}$ O% O29 OFM O O O OHHHHHrLrrcc(tj|_dSrO)rgrrf)rcs rJrr[s%GMMMrLceZdZdZdefdZddZdZdZd Z de fd Z de fd Z dd ZddZddZdZdeefdZd S)rzG Pipeline implementation for multiple logical Redis databases. rc"g|_||_dSrO)_command_stack_client)rIrs rJrKzPipeline.__init__ds  rLrIrMc K|SrOr\rQs rJrRzPipeline.__aenter__h  rLcK|d{V|j|||d{VdSrO)resetrrYrIrUrVrWs rJrYzPipeline.__aexit__ksXjjlll$$Xy)DDDDDDDDDDDrLcN|SrO) _async_self __await__rQs rJrzPipeline.__await__os !!++---rLc K|SrOr\rQs rJrzPipeline._async_selfrrrLc*t|jSrO)lenrrQs rJ__len__zPipeline.__len__us4&'''rLcdS)z1Pipeline instances should always evaluate to TrueTr\rQs rJ__bool__zPipeline.__bool__xstrLNcKg|_dSrO)rrQs rJrzPipeline.reset|s rLc>K|d{VdS)zClose the pipelineN)rrQs rJrzPipeline.acloses,jjllrLc>|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 rJpipeline_execute_commandz!Pipeline.pipeline_execute_commands$ ""D'?333 rLc|j|i|S)zAdds a command to the stack)rrIrrs rJrzPipeline.execute_commands,t,d=f===rLc@K|jjs|jd{V |jjt |jd{V |d{VS#|d{VwxYw)z0Execute all the commands in the current pipelineN)rr@rPr?execute_pipelinetuplerrrQs rJrzPipeline.executes|' ,,))++ + + + + + + + 6GGd)** **,,        $**,,        s 6BB)rIrrMrrMN)rMr)rrrrrrKrRrYrrintrrrrrrrrrrr\rLrJrr_s}EEE...((((($!!!!    >>> tCy      rLrceZdZdZdefdZddZddZdZe de fd Z d e fd Z d ed efd Zd efdZd ed efdZdZ dde deefdZddddeddfdZdS)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 Nr\)rr?r)rIrrs rJrKzPubSub.__init__s/ , %,66v66666rLrMc K|SrOr\rQs rJrRzPubSub.__aenter__rrLNc>K|d{VdSrO)rrs rJrYzPubSub.__aexit__s,kkmmrLcPK|jjdd{VS)Nrrr?execute_pubsub_methodrQs rJrz PubSub.acloses1\2HHRRRRRRRRRrLc.|jjjjSrO)rr? active_pubsub subscribedrQs rJrzPubSub.subscribeds|,:EErLrcBK|jjjdg|Rd{VS)NrrrIrs rJrzPubSub.execute_commandsPHT\2H  $          rLrcHK|jjjdg|Ri|d{VS)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()``. psubscribeNrrs rJrzPubSub.psubscribes`IT\2H     #)         rLcBK|jjjdg|Rd{VS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscribeNrrs rJrzPubSub.punsubscribesS IT\2H  !          rLcHK|jjjdg|Ri|d{VS)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()``. subscribeNrrs rJrzPubSub.subscribes`IT\2H     "(         rLcBK|jjjdg|Rd{VS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscribeNrrs rJrzPubSub.unsubscribesS IT\2H             rLFignore_subscribe_messagesrcVK|jjd||d{VS)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_message)rrNr)rIrrs rJrzPubSub.get_messagesR\2HH &?I         rLg?)exception_handler poll_timeoutrcVK|jj|||d{VS)aProcess pub/sub messages using registered callbacks. This is the equivalent of :py:meth:`redis.PubSub.run_in_thread` in redis-py, but it is a coroutine. To launch it as a separate task, use ``asyncio.create_task``: >>> task = asyncio.create_task(pubsub.run()) To shut it down, use asyncio cancellation: >>> task.cancel() >>> await task ) sleep_timerrN)rr?execute_pubsub_run)rIrrs rJrunz PubSub.runsR&\2EE#7HQUF         rL)rMrr)Fr)rrrrrrKrRrYrpropertyrrrrrr rrrrrrrrrr\rLrJrrs 7} 7 7 7 7SSSFDFFFXF :     h  -            X          SV  )- @H    &!             rLr)2rAloggingtypingrrrrrrr redis.asyncio.clientr &redis.asyncio.multidb.command_executorr redis.asyncio.multidb.configr r redis.asyncio.multidb.databaserr&redis.asyncio.multidb.failure_detectorr!redis.asyncio.multidb.healthcheckrrredis.backgroundrredis.commandsrrredis.multidb.circuitrrrgredis.multidb.exceptionrr redis.typingrrr redis.utilsr getLoggerrrrrrrr\rLrJrsLMMMMMMMMMMMMMMMMMM......IIIIIILLLLLLLLCCCCCCCCGGGGGGLLLLLLLL000000FFFFFFFF000000222222XXXXXXXX3333333333$$$$$$  8 $ $AHAHAHAHAH,.?AHAHAHH &&&&&AAAAA'):AAAHq q q q q q q q q q rL