iSddlZddlZddlmZmZmZmZmZmZddl 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.dZdZdZd Z de fd Z d e dd fd Z d/dedefdZde de fdZd e fdZd e defdZdefdZdefdZdZdZd dd dded geeeeffd!ed"e e!d#ed$e ef d%Z"d&Z#de$e%effd'Z&d(Z'd e defd)Z(d*e)d+e*d,e*fd-Z+d S)0 MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc `||_|js|n|j|_|j|_|j|_ |j s| n|j |_ |j |n|j |_|j|j|j|_|j|_|j|_|jt0gt3|j |j|j|j|j|j|j|j|_d|_t=j|_ tC|_"||_#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_policyvalue_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 n/Users/user/workspace/sujinbaek/cqa-test-app/venv/lib/python3.11/site-packages/redis/asyncio/multidb/client.py__init__zMultiDBClient.__init__)s **,,' &F ( ( * * *%  '-&B#  & , , . . ! + *F , , . . .) '/  , , . . .)  --do>>>'-'D$!'!8$2 335K4LMMM 6"5o-"5$6!0!3#'#? ! ! ! !  022 "&%)"""rMreturncLK|js|d{V|SN)rD initializerMs rN __aenter__zMultiDBClient.__aenter__Vs8 $//## # # # # # # # rPcrK|jr|j|jr|j|jD]}||jd{V|jjr+|jjj d{VdSdSrS) rJcancelrLrKr8closerCactive_databaseclientaclose)rMhc_tasks rNr\zMultiDBClient.aclose[s  " -  # * * , , ,  % 0  & - - / / /~  G NN    '--/////////  0 H'7>EEGG G G G G G G G G G H HrPc>K|d{VdSrSr\rMexc_type exc_value tracebacks rN __aexit__zMultiDBClient.__aexit__k,kkmmrPcK|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_checkrE create_taskrHrun_recurring_asyncr5_check_databases_healthrJr0circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDrC_active_databaserrD)rMis_active_db_founddatabaseweights rNrTzMultiDBClient.initializens 00222222222#*"5   2 2+,  # # # $ * * Hf   - -d.T U U U%77@R7:B%6%)"! *F  rPc|jS)zE Returns a sorted (by weight) list of all databases. )r0rUs rN get_databaseszMultiDBClient.get_databasess rPrsNcKd}|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_healthrkrnrorp get_top_nrCset_active_databaserMANUALr)rMrsexists existing_db_highest_weighted_dbs rNr}z!MultiDBClient.set_active_databases "o  NKh&&' PNOO O##H---------   !W^ 3 3%)_%>%>q%A%A!%D " ';;+2        F& ?   rPTskip_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)r[rkrthealth_check_urlry) client_kwargsupdaterrfrom_urlrI client_class from_pool set_retryrkdefault_circuit_breakerrrtrr{rr0r|add_change_active_database)rMr&rr[rkrsrhighest_weights rN 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{VdSdSdSrS) rtrkrnrorprCr}r AUTOMATIC)rMrrs rNrz%MultiDBClient._change_active_databases  "9"@ @ @$*gn<<';;/9          A @< Adds a new failure detector to the database. N)r:append)rMrs rNadd_failure_detectorz"MultiDBClient.add_failure_detectors" &&'788888rP healthcheckcK|j4d{V|j|dddd{VdS#1d{VswxYwYdS)z: Adds a new health check to the database. N)rGr3r)rMrs rNadd_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)rDrTrCexecute_commandrMargsoptionss rNrzMultiDBClient.execute_commandsb $//## # # # # # # #:T*:DLGLLLLLLLLLrPc t|S)z: Enters into pipeline mode of the client. )PipelinerUs rNpipelinezMultiDBClient.pipeline's~~rPF shard_hintvalue_from_callable watch_delayfuncrwatchesrrrcK|js|d{V|jj|g|R|||dd{VS)z3 Executes callable as transaction. Nr)rDrTrCexecute_transaction)rMrrrrrs rN transactionzMultiDBClient.transaction-s $//## # # # # # # #>T*>    " 3#           rPcbK|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)rDrTPubSub)rMkwargss rNpubsubzMultiDBClient.pubsubCsK  $//## # # # # # # #d%%f%%%rPc"Kig|_|jD]K\}}tj||}||<|j|Ltj|jddid{V}fdt|j|D}|D]]\}}t|trC|j }tj |j_t d|jd||<^|S)zk Runs health checks as a recurring task. Runs health checks against all databases. return_exceptionsTNc(i|]\}}||Srr).0taskresult task_to_dbs rN z9MultiDBClient._check_databases_health..^s1   )5vJt f   rPz%Health check failed, due to exception)exc_infoF)rKr0rErhr{rgatherzipitems isinstancerrsroOPENrkrnloggerdebugoriginal_exception) rMrsrrresults db_resultsr unhealthy_dbrs @rNrjz%MultiDBClient._check_databases_healthNsG 46 ? ( (KHa&t'<'%??@  SM  "  e_     , & & & tHdN/C    D0}$L%L29LFMLLLLLLrPr%rkc(tj|_dSrS)rorrn)rks rNrrs%GMMMrPceZdZdZdefdZddZdZdZd Z de fd Z de fd Z dd ZddZddZdZdeefdZd S)rzG Pipeline implementation for multiple logical Redis databases. r[c"g|_||_dSrS)_command_stack_client)rMr[s rNrOzPipeline.__init__s  rPrMrQc K|SrSrrUs rNrVzPipeline.__aenter__  rPcK|d{V|j|||d{VdSrS)resetrrdr`s rNrdzPipeline.__aexit__sXjjlll$$Xy)DDDDDDDDDDDrPcN|SrS) _async_self __await__rUs rNrzPipeline.__await__s !!++---rPc K|SrSrrUs rNrzPipeline._async_selfrrPc*t|jSrS)rrrUs rN__len__zPipeline.__len__s4&'''rPcdS)z1Pipeline instances should always evaluate to TrueTrrUs rN__bool__zPipeline.__bool__strPNcKg|_dSrS)rrUs rNrzPipeline.resets rPc>K|d{VdS)zClose the pipelineN)rrUs rNr\zPipeline.acloses,jjllrPc>|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 rNpipeline_execute_commandz!Pipeline.pipeline_execute_commands$ ""D'?333 rPc|j|i|S)zAdds a command to the stack)rrMrrs rNrzPipeline.execute_commands,t,d=f===rPc@K|jjs|jd{V |jjt |jd{V |d{VS#|d{VwxYw)z0Execute all the commands in the current pipelineN)rrDrTrCexecute_pipelinetuplerrrUs rNrzPipeline.executes|' ,,))++ + + + + + + + 6GGd)** **,,        $**,,        s 6BB)rMrrQrrQN)rQr)rrrrr%rOrVrdrrintrrrrr\rrrrrrrPrNrrs}EEE...((((($!!!!    >>> tCy      rPrceZdZdZdefdZddZddZdZe de fd Z d e fd Z d ed efd Zd efdZd ed efdZdZ dde deefdZddddeddfdZdS)rz2 PubSub object for multi database client. r[c 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)rrCr)rMr[rs rNrOzPubSub.__init__s/ , %,66v66666rPrQc K|SrSrrUs rNrVzPubSub.__aenter__rrPNc>K|d{VdSrSr_r`s rNrdzPubSub.__aexit__ rerPcPK|jjdd{VS)Nr\rrCexecute_pubsub_methodrUs rNr\z PubSub.aclose s1\2HHRRRRRRRRRrPc.|jjjjSrS)rrC active_pubsub subscribedrUs rNrzPubSub.subscribeds|,:EErPrcBK|jjjdg|Rd{VS)NrrrMrs rNrzPubSub.execute_commandsPHT\2H  $          rPrcHK|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 rNrzPubSub.psubscribes`IT\2H     #)         rPcBK|jjjdg|Rd{VS)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscribeNrrs rNrzPubSub.punsubscribe%sS IT\2H  !          rPcHK|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 rNrzPubSub.subscribe.s`IT\2H     "(         rPcBK|jjjdg|Rd{VS)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscribeNrrs rNrzPubSub.unsubscribe:sS IT\2H             rPFignore_subscribe_messagestimeoutcVK|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)rr Nr)rMrr s rNr"zPubSub.get_messageCsR\2HH &?I         rPg?)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)rrCexecute_pubsub_run)rMr#r$s rNrunz PubSub.runSsR&\2EE#7HQUF         rP)rQrr)Fr)rrrrr%rOrVrdr\propertyrrr!rr r rrrrrrrr"r(rrPrNrrs 7} 7 7 7 7SSSFDFFFXF :     h  -            X          SV  )- @H    &!             rPr);rEloggingtypingrrrrrrredis.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.circuitrrroredis.multidb.exceptionrrrredis.observability.attributesr redis.typingr r!r" redis.utilsr# getLoggerrrr%rrrrrPrNr<sBBBBBBBBBBBBBBBB......IIIIII NMMMMMMMMMGGGGGGLLLLLLLL%%%%%%000000######FFFFFFFF000000222222 =<<<<<3333333333$$$$$$  8 $ $ILILILILIL,.?ILILILX &&&&&AAAAA'):AAAHq q q q q q q q q q rP