/ iTddlZddlZddlmZmZmZmZmZmZddl m Z ddl m Z ddl mZmZmZmZddlmZmZmZddlmZddlmZmZdd lmZdd lmZdd lm Z dd l!m"Z"m#Z#dd l$m%Z%ddl$m&Z'ddl(m)Z)m*Z*m+Z+ddl,m-Z-ddl.m/Z/m0Z0m1Z1ddl2m3Z3ej4e5Z6e3Gdde#e"Z7de%fdZ8Gdde#e"Z9GddZ:dS)N)Any AwaitableCallableListOptionalUnion) PubSubHandler)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheck MultiDbConfig) AsyncDatabaseDatabase Databases)AsyncFailureDetector) HealthCheckHealthCheckPolicy)Retry)BackgroundScheduler) NoBackoff)AsyncCoreCommandsAsyncRedisModuleCommands)CircuitBreaker)State)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)GeoFailoverReason)ChannelT EncodableTKeyT) experimentalc peZdZdZdefdZd.dZdZdZde fd Z d e dd fd Z d/de defdZde de fdZd e fdZd e defdZdefdZdefdZdZdZd dd ddedgeeeeffd ed!ee d"ed#eef d$Z!d%Z"de#e$effd&Z%d'Z&d e defd(Z'd)e(d*e)d+e)fd,Z*d-Z+d S)0 MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover 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 C:\Users\Dell Inspiron 16\Desktop\tws\AgrotaPowerBi\back-agrota-powerbi\mcp-client-agrota\venv\Lib\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 "&%)"""rOreturncLK|js|d{V|SN)rF initializerOs rP __aenter__zMultiDBClient.__aenter__Ws8 $//## # # # # # # # rRcK|jr|j|jr|j|jD]}|dSrU)rLcancelrNrM)rOexc_type exc_value tracebackhc_tasks rP __aexit__zMultiDBClient.__aexit__\st  " -  # * * , , ,  % 0  & - - / / /~  G NN      rRcK|d{Vtj|j|j|j|_d}|jD]N\}}|j |j |j j tjkr|s||j_d}O|st#dd|_dS)zT Perform initialization of databases to define their initial state. NFTz4Initial connection failed - no active database found)_perform_initial_health_checkrG create_taskrJrun_recurring_asyncr5_check_databases_healthrLr0circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDrE_active_databaserrF)rOis_active_db_founddatabaseweights rPrVzMultiDBClient.initializeds 00222222222#*"5   2 2+,  # # # $ * * Hf   - -d.T U U U%77@R7:B%6%)"! *F  rRc|jS)zE Returns a sorted (by weight) list of all databases. )r0rWs rP get_databaseszMultiDBClient.get_databasess rRrmNcKd}|jD]\}}||krd}n|std||d{V|jjt jkrP|jdd\}}|j |tj 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) r0 ValueError_check_db_healthrerhrirj get_top_nrEset_active_databaserMANUALr)rOrmexists existing_db_highest_weighted_dbs rPrwz!MultiDBClient.set_active_databases "o  NKh&&' PNOO O##H---------   !W^ 3 3%)_%>%>q%A%A!%D " ';;+2        F& ?   rRTskip_initial_health_checkcZK|jdtdti|jr#|jjj|jfi|j}ny|jr[|jtdt|jj|j}n|jjdi|j}|j | n|j }t|||j |j } ||d{Vn#t$r|sYnwxYw|jdd\}}|j||j |||d{VdS) 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. retryr)retriesbackoff)connection_poolN)clientrernhealth_check_urlrs) client_kwargsupdaterrfrom_urlrK client_class from_pool set_retryredefault_circuit_breakerrrnrrurr0rvadd_change_active_database)rOr&r}rrermr|highest_weights rP add_databasezMultiDBClient.add_databases ##WeAy{{.S.S.S$TUUU ? G7T\.7#)#7FF  G   & &uQ 'L'L'L M M M\.88 & 09FF/T\.FF1EFFF~%  * * , , ,  =#4     ''11 1 1 1 1 1 1 1 1)   ,    /3o.G.G.J.J1.M+^ Hho666**85HIIIIIIIIIIIsD99 E E  new_databasehighest_weight_databasecK|j|jkrG|jjtjkr/|j|tjd{VdSdSdSrU) rnrerhrirjrErwr AUTOMATIC)rOrrs rPrz%MultiDBClient._change_active_databases  "9"@ @ @$*gn<<';;/9          A @< Adds a new failure detector to the database. N)r<append)rOrs rPadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rR healthcheckcK|j4d{V|j|dddd{VdS#1d{VswxYwYdS)z: Adds a new health check to the database. N)rIr3r)rOrs rPadd_health_checkzMultiDBClient.add_health_check s= 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)rFrVrEexecute_commandrOargsoptionss rPrzMultiDBClient.execute_commandsb $//## # # # # # # #:T*:DLGLLLLLLLLLrRc t|S)z: Enters into pipeline mode of the client. )PipelinerWs rPpipelinezMultiDBClient.pipelines~~rRF shard_hintvalue_from_callable watch_delayfuncrwatchesrrrcK|js|d{V|jj|g|R|||dd{VS)z3 Executes callable as transaction. Nr)rFrVrEexecute_transaction)rOrrrrrs rP transactionzMultiDBClient.transaction#s $//## # # # # # # #>T*>    " 3#           rRcbK|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)rFrVPubSub)rOkwargss rPpubsubzMultiDBClient.pubsub9sK  $//## # # # # # # #d%%f%%%rRcK ig|_|jD]K\}}tj||}||<|j|Ltjtj|jddi|jd{V}n'#tj $rtj dwxYwfdt|j|D}| D]x\}}t|trD|j}tj|j_t&d|jd ||<^t|t,rd ||<y|S) zk Runs health checks as a recurring task. Runs health checks against all databases. return_exceptionsT)timeoutNz4Health check execution exceeds health_check_intervalc(i|]\}}||Srr).0taskresult task_to_dbs rP z9MultiDBClient._check_databases_health..\s1   )5vJt f   rRz%Health check failed, due to exception)exc_infoF)rMr0rGrbrurwait_forgatherr5 TimeoutErrorzipitems isinstancerrmriOPENrerhloggerdebugoriginal_exception Exception) rOrmr{rresults db_resultsr unhealthy_dbrs @rPrdz%MultiDBClient._check_databases_healthDs  79JDN# , , !*4+@+@+J+JKK#+ 4 %%d++++#,G$GG3GG#   &F       9EEGG G G G G G G G G G H HrR)rOr%rSr%)T),__name__ __module__ __qualname____doc__rrQrXr_rVrrprrwr boolrrrfloatrrrrrrrrrrrr"rstrrrdictrrdrarurrirgrrrRrPr%r%"s ,*},*,*,*,*\ " " " Hy  - D    8IM/J/J$/JAE/J/J/J/Jb ) DQ     m    J]JEJJJJ&95I9999 4+4444MMM%)$)'+     |U3 #+>%??@  SM  "  e_     , & & &*tHdN/C****X0}$L%L29LFMLLLL(HHHHHrRr%rec(tj|_dSrU)rirrh)res rPrrs%GMMMrRceZdZdZdefdZddZdZdZd Z de fd Z de fd Z dd ZddZddZdZdeefdZd S)rzG Pipeline implementation for multiple logical Redis databases. rc"g|_||_dSrU)_command_stack_client)rOrs rPrQzPipeline.__init__s  rRrOrSc K|SrUrrWs rPrXzPipeline.__aenter__  rRcK|d{V|j|||d{VdSrU)resetrr_rOr[r\r]s rPr_zPipeline.__aexit__sXjjlll$$Xy)DDDDDDDDDDDrRcN|SrU) _async_self __await__rWs rPrzPipeline.__await__s !!++---rRc K|SrUrrWs rPrzPipeline._async_selfrrRc*t|jSrU)rrrWs rP__len__zPipeline.__len__s4&'''rRcdS)z1Pipeline instances should always evaluate to TrueTrrWs rP__bool__zPipeline.__bool__strRNcKg|_dSrU)rrWs rPrzPipeline.resets rRc>K|d{VdS)zClose the pipelineN)rrWs rPrzPipeline.acloses,jjllrRc>|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 rPpipeline_execute_commandz!Pipeline.pipeline_execute_commands$ ""D'?333 rRc|j|i|S)zAdds a command to the stack)rrOrrs rPrzPipeline.execute_commands,t,d=f===rRc@K|jjs|jd{V |jjt |jd{V |d{VS#|d{VwxYw)z0Execute all the commands in the current pipelineN)rrFrVrEexecute_pipelinetuplerrrWs rPrzPipeline.executes|' ,,))++ + + + + + + + 6GGd)** **,,        $**,,        s 6BB)rOrrSrrSN)rSr)rrrrr%rQrXr_rrintrrrrrrrrrrrrRrPrrs}EEE...((((($!!!!    >>> tCy      rRrceZdZdZdefdZddZddZdZe 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)rrEr)rOrrs rPrQzPubSub.__init__s/ , %,66v66666rRrSc K|SrUrrWs rPrXzPubSub.__aenter__ rrRNc>K|d{VdSrU)rrs rPr_zPubSub.__aexit__s,kkmmrRcPK|jjdd{VS)NrrrEexecute_pubsub_methodrWs rPrz PubSub.acloses1\2HHRRRRRRRRRrRc.|jjjjSrU)rrE active_pubsub subscribedrWs rPrzPubSub.subscribeds|,:EErRrcBK|jjjdg|Rd{VS)NrrrOrs rPrzPubSub.execute_commandsPHT\2H  $          rRrcHK|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 rPrzPubSub.psubscribes`IT\2H     #)         rRcBK|jjjdg|Rd{VS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscribeNrrs rPrzPubSub.punsubscribe)sS IT\2H  !          rRcHK|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 rPrzPubSub.subscribe2s`IT\2H     "(         rRcBK|jjjdg|Rd{VS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscribeNrrs rPr zPubSub.unsubscribe>sS IT\2H             rRFignore_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)r"rNr)rOr"rs rPr$zPubSub.get_messageGsR\2HH &?I         rRg?)exception_handler poll_timeoutr&cVK|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_timer%rN)rrEexecute_pubsub_run)rOr%r&s rPrunz PubSub.runWsR&\2EE#7HQUF         rR)rSrr )Fr!)rrrrr%rQrXr_rpropertyrrr!rr r rrrrr rrr$r*rrRrPrrs 7} 7 7 7 7SSSFDFFFXF :     h  -            X          SV  )- @H    &!             rRr);rGloggingtypingrrrrrrredis.asyncio.clientr &redis.asyncio.multidb.command_executorr redis.asyncio.multidb.configr r r rredis.asyncio.multidb.databaserrr&redis.asyncio.multidb.failure_detectorr!redis.asyncio.multidb.healthcheckrrredis.asyncio.retryrredis.backgroundr redis.backoffrredis.commandsrrredis.multidb.circuitrrriredis.multidb.exceptionrrrredis.observability.attributesr redis.typingr r!r" redis.utilsr# getLoggerrrr%rrrrrRrPr>sBBBBBBBBBBBBBBBB......IIIIII NMMMMMMMMMGGGGGGLLLLLLLL%%%%%%000000######FFFFFFFF000000222222 =<<<<<3333333333$$$$$$  8 $ $MHMHMHMHMH,.?MHMHMH` &&&&&AAAAA'):AAAHq q q q q q q q q q rR