Ki}YddlZddlZddlZddlmZmZmZmZddlm Z m Z ddl m Z ddl mZddlmZddlmZmZddlmZdd lmZdd lmZdd lmZdd lmZmZmZm Z dd l!m"Z"m#Z#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,ddl-m.Z.ddl/m0Z0ejbe2Z3e0GddeeZ4defdZ5GddeeZ6GddZ7y)N)AnyCallableListOptional) HealthCheckHealthCheckPolicy)BackgroundScheduler) NoBackoff)PubSubWorkerThread) CoreCommandsRedisModuleCommands)MaintNotificationsConfig)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheck MultiDbConfig)Database Databases SyncDatabase)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector)GeoFailoverReason)Retry) experimentalc eZdZdZdefdZdZdZdefdZ de dd fd Z d%de d e fd Zd e de fdZdefdZde defdZdefdZdefdZdZdZdedgd ffdZdZde de fdZdeee ffdZdZ d e!d!e"d"e"fd#Z#d$Z$y )& MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc |j|_|js|jn |j|_|j |_|jj|_ |js|jn |j|_ |j|jn |j|_|jj!|j|j"|_|j&|_|j*|_|j,j/t0ft3|j|j|j,|j|j4|j6|j(|j$|_d|_t=|_tAjB|_"||_#y)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_errorsConnectionRefusedErrorrr(r)command_executor initializedr _bg_scheduler threadingLock_hc_lock_config)selfr"s `/home/jay/workspace/scripts/.codegraph-venv/lib/python3.12/site-packages/redis/multidb/client.py__init__zMultiDBClient.__init__*s **,''  ( ( *%%  '-&B&B#  & & , , . ! ++  , , .)) ''/  , , .))  --doo>'-'D'D$!'!8!8$22 335K4MN 6"55oo--"55$66!00!33#'#?#? ! !02!(  cD |jy#t$rYywxYwN)close ExceptionrFs rG__del__zMultiDBClient.__del__Ts$  JJL      c|jj|j|jj|j|j d}|j D]h\}}|jj|j|jjtjk(sS|rV||j_d}j|s tdd|_y)zT Perform initialization of databases to define their initial state. FTz4Initial connection failed - no active database foundN)rA run_coro_sync_perform_initial_health_checkrun_recurring_coror1_check_databases_healthr,circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr?_active_databaserr@)rFis_active_db_founddatabaseweights rG initializezMultiDBClient.initialize]s (()K)KL --  ' '  ( ( # $ * Hf    - -d.T.T U%%7@R:B%%6%)" *"*F  rIreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r,rNs rG get_databaseszMultiDBClient.get_databasessrIr^Ncd}|jD]\}}||k(s d}n|s td|jj|j||j j tjk(rC|jjdd\}}|tjf|j_ ytd)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, ValueErrorrArR_check_db_healthrVrYrZr[ get_top_nrMANUALr?active_databaser)rFr^exists existing_db_highest_weighted_dbs rGset_active_databasez!MultiDBClient.set_active_databases"oo NKh&  NO O (()>)>I    ! !W^^ 3%)__%>%>q%A!%D " !((5D ! ! 1 & ?  rIskip_initial_health_checkctdt|jd<d|jvrtd|jd<|jr<|j j j|jfi|j}n|jr_|jjtdt|j j j|j}n&|j j d i|j}|j|jn |j}t|||j|j } |jj|j ||j$j'd d\}}|j$j)||j|j+||y#t"$r|sYhwxYw) 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. r)retriesbackoffretrymaint_notifications_configF)enabled)connection_poolN)clientrVr_health_check_urlrf)rr client_kwargsrfrom_urlrE client_class from_pool set_retryrVdefault_circuit_breakerrr_rzrArRrhrr,riadd_change_active_database)rFr"rqryrVr^rohighest_weights rG add_databasezMultiDBClient.add_databases).a(MW% (v/C/C C(7  != > ??7T\\..77#)#7#7F      & &uQ 'L M\\..88 & 0 09F/T\\..F1E1EFF~~%  * * ,  ==#44       , ,T-B-BH M /3oo.G.G.J1.M+^ Hhoo6 $$X/BC * ,- s/&G// G>=G> new_databasehighest_weight_databasec|j|jkDrJ|jjtjk(r"|t j f|j_yyyrK) r_rVrYrZr[r AUTOMATICr?rk)rFrrs rGrz%MultiDBClient._change_active_databasesZ   "9"@"@ @$$**gnn<!++5D ! ! 1= ArIc|jj|}|jjdd\}}||krJ|jjt j k(r"|tjf|j_ yyy)z< Removes a database from the database list. rfrN) r,removerirVrYrZr[rrjr?rk)rFr^r_rors rGremove_databasezMultiDBClient.remove_databases''1.2oo.G.G.J1.M+^ f $#++11W^^C$!((5D ! ! 1D %rIr_cd}|jD]\}}||k(s d}n|s td|jjdd\}}|jj||||_|j ||y)z< Updates a database from the database list. NTrerfr)r,rgri update_weightr_r)rFr^r_rlrmrnrors rGupdate_database_weightz$MultiDBClient.update_database_weights"oo NKh&  NO O.2oo.G.G.J1.M+^ %%h7  $$X/BCrIfailure_detectorc:|jj|y)z> Adds a new failure detector to the database. N)r6append)rFrs rGadd_failure_detectorz"MultiDBClient.add_failure_detector s &&'78rI healthcheckc||j5|jj|dddy#1swYyxYw)z: Adds a new health check to the database. N)rDr/r)rFrs rGadd_health_checkzMultiDBClient.add_health_checks4]] 4    & &{ 3 4 4 4s2;cr|js|j|jj|i|S)zB Executes a single command and return its result. )r@r`r?execute_commandrFargsoptionss rGrzMultiDBClient.execute_commands5 OO 4t$$44dFgFFrIct|S)z: Enters into pipeline mode of the client. )PipelinerNs rGpipelinezMultiDBClient.pipeline"s~rIfuncrcx|js|j|jj|g||S)z3 Executes callable as transaction. )r@r`r?execute_transaction)rFrwatchesrs rG transactionzMultiDBClient.transaction(s: OO 8t$$88RR'RRrIc R|js|jt|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)rFkwargss rGpubsubzMultiDBClient.pubsub1s'  OO d%f%%rIcK|j5t|j}ddd|jj |d{}|sH|j j tjk7rtj|j _|S|rF|j j tjk7rtj|j _|S#1swYxYw7w)zO Runs health checks on the given database until first failure. N) rDlistr/r4executerVrYrZOPENr[)rFr^r- is_healthys rGrhzMultiDBClient._check_db_health<s]] 6 !4!45M 6 44<<]HUU %%5)0  &  H,,22gnnD%,^^H   " 6 6Vs( C3C%'C3 C1 BC3%C.*C3czKi}g|_|jD]I\}}tj|j |}|||<|jj |Ktj |jddid{}t|j|Dcic] \}}||| }}}|jD]g\}}t|ts|j}tj|j_t j#d|j$d||<i|S7cc}}ww)zk Runs health checks as a recurring task. Runs health checks against all databases. return_exceptionsTNz%Health check failed, due to exception)exc_infoF) _hc_tasksr,asyncio create_taskrhrgatherzipitems isinstancerr^rZrrVrYloggerdebugoriginal_exception) rF task_to_dbr^rntaskresultsresult db_results unhealthy_dbs rGrUz%MultiDBClient._check_databases_healthOs4 46 ?? (KHa&&t'<' > @ 1   s/B B)(B))T)%__name__ __module__ __qualname____doc__rrHrOr`rrcrrprboolrrrrfloatrrrrrrrrrrrhdictrUrSrrZrXrLr{rIrGr!r!#s, (}(T$ Ly  L T :IM6D$6DAE6Dp ( CO  D|DUD&9_9 4K4G S*t); <S &|& tHdN/C D0L%L29LFML*ArIr!rVc.tj|_yrK)rZrrY)rVs rGrrs%%GMrIcxeZdZdZdefdZddZdZdZde fdZ de fd Z dd Z dd Zdd ZdZdeefdZy )rzG Pipeline implementation for multiple logical Redis databases. ryc g|_||_yrK)_command_stack_client)rFrys rGrHzPipeline.__init__s  rIrac|SrKr{rNs rG __enter__zPipeline.__enter__ rIc$|jyrKreset)rFexc_type exc_value tracebacks rG__exit__zPipeline.__exit__  rIcD |jy#t$rYywxYwrKrrMrNs rGrOzPipeline.__del__s"  JJL   rPc,t|jSrK)rrrNs rG__len__zPipeline.__len__s4&&''rIcy)z1Pipeline instances should always evaluate to TrueTr{rNs rG__bool__zPipeline.__bool__srINcg|_yrK)rrNs rGrzPipeline.resets  rIc$|jy)zClose the pipelineNrrNs rGrLzPipeline.closes  rIc@|jj||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 rGpipeline_execute_commandz!Pipeline.pipeline_execute_commands! ""D'?3 rIc&|j|i|S)zAdds a command to the stack)rrFrrs rGrzPipeline.execute_commands,t,,d=f==rIc|jjs|jj |jjj t |j |jS#|jwxYw)z0Execute all the commands in the current pipeline)rr@r`r?execute_pipelinetuplerrrNs rGrzPipeline.executes_||'' LL # # % <<00AAd))* JJLDJJLs 7A::B )rarraN)rrrrr!rHrrrOintrrrrrLrrrrrr{rIrGrrs^} (($! > c rIrc eZdZdZdefdZddZddZddZdd Z e de fd Z d Z d Zd ZdZdZdZdZ dde defdZ dde defdZ ddede deede ddf dZy) rz2 PubSub object for multi database client. ryc ^||_|jjjdi|y)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)rFryrs rGrHzPubSub.__init__s( , %%,,6v6rIrac|SrKr{rNs rGrzPubSub.__enter__rrINcD |jy#t$rYywxYwrKrrNs rGrOzPubSub.__del__ s$  JJL   rPcL|jjjdS)Nrrr?execute_pubsub_methodrNs rGrz PubSub.resets||,,BB7KKrIc$|jyrKrrNs rGrLz PubSub.closerrIcV|jjjjSrK)rr? active_pubsub subscribedrNs rGrzPubSub.subscribeds||,,::EEErIcP|jjjdg|S)NrrrFrs rGrzPubSub.execute_commands,Bt||,,BB  $  rIcV|jjjdg|i|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 rGr zPubSub.psubscribe#7Ct||,,BB   #)  rIcP|jjjdg|S)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rGr zPubSub.punsubscribe// Ct||,,BB  !  rIcV|jjjdg|i|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 rGrzPubSub.subscribe8s7Ct||,,BB   "(  rIcP|jjjdg|S)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rGrzPubSub.unsubscribeDs( Ct||,,BB=XSWXXrIcV|jjjdg|i|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 rGrzPubSub.ssubscribeKr rIcP|jjjdg|S)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rGrzPubSub.sunsubscribeWrrIignore_subscribe_messagestimeoutcR|jjjd||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_messagerrrrFrrs rGrzPubSub.get_message`s0||,,BB &?C  rIcR|jjjd||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 rGrzPubSub.get_sharded_messageps0||,,BB !&?C  rI sleep_timedaemonexception_handlersharded_pubsubr cV|jjj|||||S)N)r r!rr")rr?execute_pubsub_run)rFrr r!r"s rG run_in_threadzPubSub.run_in_threads6||,,?? /) @  rI)rarr)F)r&FNF)rrrrr!rHrrOrrLpropertyrrrr r rrrrrrrrrr%r{rIrGrrs 7} 7LFDFF     Y   IL )- @E "IL )- @E $ 04$      $H-      rIr)8rloggingrBtypingrrrr!redis.asyncio.multidb.healthcheckrrredis.backgroundr redis.backoffr redis.clientr redis.commandsr r redis.maint_notificationsrredis.multidb.circuitrrrZredis.multidb.command_executorrredis.multidb.configrrrrredis.multidb.databaserrrredis.multidb.exceptionrrrredis.multidb.failure_detectorrredis.observability.attributesr redis.retryr redis.utilsr getLoggerrrr!rrrr{rIrGr:s00L0#+<>02A ED ;<$   8 $JA'JAJAZ &&@"L@FU U rI