o iU@svddlZddlZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZddlmZddlmZmZdd lmZdd lmZdd lmZdd lmZdd lmZmZmZm Z ddl!m"Z"m#Z#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-ddl.m/Z/ddl0m1Z1ddl2m3Z3e4e5Z6e3GdddeeZ7defddZ8GdddeeZ9GdddZ:dS)N) as_completed)ThreadPoolExecutor)AnyCallableListOptional)BackgroundScheduler) NoBackoff)PubSubWorkerThread) CoreCommandsRedisModuleCommands)MaintNotificationsConfig)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheck MultiDbConfig)Database Databases SyncDatabase)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector) HealthCheckHealthCheckPolicy)GeoFailoverReason)Retry) experimentalc@s(eZdZdZdefddZddZdefdd Zd e dd fd d Z d8de de fddZ de de fddZd efddZd e defddZdefddZdefdd Zd!d"Zd#d$Zd%ed&gd ffd'd(Zd)d*Zd e de fd+d,Zdeee ffd-d.Zd/d0Zd1e d2e!d3e!fd4d5Z"d6d7Z#d S)9 MultiDBClientz Client that operates on multiple logical Redis databases. Should be used in Client-side geographic failover database setups. configc Cs||_|js |n|j|_|j|_|j|j |j |_ |j s%| n|j |_|jdur2|n|j|_|j|j|j|_|j|_|j|_|jtft|j|j|j|j|j|j|j|jd|_d|_t !|_"t#|_$||_%dS)N)failure_detectors databases command_retryfailover_strategyfailover_attemptsfailover_delayevent_dispatcherauto_fallback_intervalF)&r$ _databasesZ health_checksZdefault_health_checks_health_checksZhealth_check_interval_health_check_intervalZhealth_check_policyvalueZhealth_check_probesZhealth_check_probes_delay_health_check_policyr#Zdefault_failure_detectors_failure_detectorsr&Zdefault_failover_strategyZ_failover_strategyZ set_databasesr*Z_auto_fallback_intervalr)Z_event_dispatcherr%Z_command_retryZupdate_supported_errorsConnectionRefusedErrorrr'r(command_executor initialized threadingRLock_hc_lockr _bg_scheduler_config)selfr"r:K/home/app/Keep/.python/lib/python3.10/site-packages/redis/multidb/client.py__init__+sH      zMultiDBClient.__init__cCsp||j|j|jd}|jD]\}}|j|j|jj t j kr,|s,||j _ d}q|s3tdd|_dS)zT Perform initialization of databases to define their initial state. FTz4Initial connection failed - no active database foundN)_perform_initial_health_checkr7Z run_recurringr-_check_databases_healthr+circuitZon_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr2Z_active_databaserr3)r9Zis_active_db_founddatabaseweightr:r:r; initializeUs" zMultiDBClient.initializereturncCs|jS)zE Returns a sorted (by weight) list of all databases. )r+r9r:r:r; get_databaseswszMultiDBClient.get_databasesrDNcCsvd}|jD] \}}||krd}nq|std|||jjtjkr7|jdd\}}|tj f|j _ dSt d)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_healthr?rArBrC get_top_nrMANUALr2active_databaser)r9rDexists existing_db_highest_weighted_dbr:r:r;set_active_database}s$ z!MultiDBClient.set_active_databaseTskip_initial_health_checkcCstdtd|jd<d|jvrtdd|jd<|jr(|jjj|jfi|j}n"|jr@|jtdtd|jjj|jd}n |jjd i|j}|j durS| n|j }t |||j |j d }z||Wn tys|sqYnw|jd d\}}|j||j |||dS) 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)retriesbackoffretryZmaint_notifications_configF)enabled)Zconnection_poolN)clientr?rEhealth_check_urlrKr:)rr Z client_kwargsr Zfrom_urlr8Z client_classZ from_poolZ set_retryr?Zdefault_circuit_breakerrrEr\rMrr+rNadd_change_active_database)r9r"rVr[r?rDrThighest_weightr:r:r; add_databasesH   zMultiDBClient.add_database new_databasehighest_weight_databasecCs4|j|jkr|jjtjkr|tjf|j_dSdSdSN) rEr?rArBrCrZ AUTOMATICr2rP)r9rarbr:r:r;r^s  z%MultiDBClient._change_active_databasecCsP|j|}|jdd\}}||kr$|jjtjkr&|tjf|j _ dSdSdS)z< Removes a database from the database list. rKrN) r+removerNr?rArBrCrrOr2rP)r9rDrErTr_r:r:r;remove_databases  zMultiDBClient.remove_databaserEcCshd}|jD] \}}||krd}nq|std|jdd\}}|j||||_|||dS)z< Updates a database from the database list. NTrJrKr)r+rLrNZ update_weightrEr^)r9rDrErQrRrSrTr_r:r:r;update_database_weightsz$MultiDBClient.update_database_weightfailure_detectorcCs|j|dS)z> Adds a new failure detector to the database. N)r0append)r9rgr:r:r;add_failure_detectorsz"MultiDBClient.add_failure_detector healthcheckcCs8|j|j|WddS1swYdS)z: Adds a new health check to the database. N)r6r,rh)r9rjr:r:r;add_health_checks"zMultiDBClient.add_health_checkcOs |js||jj|i|S)zB Executes a single command and return its result. )r3rFr2execute_commandr9argsoptionsr:r:r;rl szMultiDBClient.execute_commandcCst|S)z: Enters into pipeline mode of the client. )PipelinerHr:r:r;pipelineszMultiDBClient.pipelinefuncrpcOs&|js||jj|g||RS)z3 Executes callable as transaction. )r3rFr2Zexecute_transaction)r9rrZwatchesror:r:r; transactionszMultiDBClient.transactioncKs|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. )r3rFPubSub)r9kwargsr:r:r;pubsub%szMultiDBClient.pubsubcCsP|j|j|}|s|jjtjkrtj|j_|S|r&|jjtjkr&tj|j_|S)zO Runs health checks on the given database until first failure. )r/executer,r?rArBOPENrC)r9rD is_healthyr:r:r;rM0s  zMultiDBClient._check_db_healthc sttjd`fddjD}i}z>t|jdD]4}z ||}|||<WqtyR}z|j}tj |j _ t j d|jdd||<WYd}~qd}~wwWn ty_td wWd|S1skwY|S) zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workerscs i|] \}}j||qSr:)submitrM).0rDrSexecutorr9r:r; Gsz9MultiDBClient._check_databases_health..)timeoutz%Health check failed, due to exception)exc_infoFNz4Health check execution exceeds health_check_interval)rlenr+rr-resultrrDrBrxr?rAloggerdebugZoriginal_exception TimeoutError)r9futuresresultsfuturerDeZ unhealthy_dbr:r}r;r>@sB     z%MultiDBClient._check_databases_healthcCs|}d}|jjtjkrd|v}n!|jjtjkr(t|t|dk}n |jjtj kr5d|v}|s@t d|jjdS)zj Runs initial health check and evaluate healthiness based on initial_health_check_policy. TFz:Initial health check failed. Initial health check policy: N) r>r8Zinitial_health_check_policyrZ ALL_AVAILABLEvaluesZMAJORITY_AVAILABLEsumrZ ONE_AVAILABLEr)r9rryr:r:r;r=es  z+MultiDBClient._perform_initial_health_checkr? old_state new_statecCs|tjkr ||jdS|tjkr)|tjkr)td|jd|j t t ||tjkr?|tjkrAt d|jddSdSdS)Nz Database z- is unreachable. Failover has been initiated.z is reachable again.) rB HALF_OPENrMrDrCrxrwarningr7Zrun_oncer_half_open_circuitinfo)r9r?rrr:r:r;r@}s   z/MultiDBClient._on_circuit_state_change_callbackcCs.|jr|j|jjr|jjjdSdS)z: Closes the client and all its resources. N)r7stopr2rPr[closerHr:r:r;rs  zMultiDBClient.close)T)$__name__ __module__ __qualname____doc__rr<rFrrIrrUrboolr`r^rrefloatrfrrirrkrlrqrrsrvrMdictr>r=rrBr@rr:r:r:r;r!$sH*" 8     % r!r?cCs tj|_dSrc)rBrrA)r?r:r:r;r rc@seZdZdZdefddZdddZdd Zd d Zde fd d Z de fddZ dddZ dddZdddZddZdeefddZdS)rpzG Pipeline implementation for multiple logical Redis databases. r[cCsg|_||_dSrc)_command_stack_client)r9r[r:r:r;r<s zPipeline.__init__rGcC|Srcr:rHr:r:r; __enter__zPipeline.__enter__cC |dSrcreset)r9exc_type exc_value tracebackr:r:r;__exit__rzPipeline.__exit__cC$z|WdStyYdSwrcr ExceptionrHr:r:r;__del__s  zPipeline.__del__cCs t|jSrc)rrrHr:r:r;__len__ zPipeline.__len__cCsdS)z1Pipeline instances should always evaluate to TrueTr:rHr:r:r;__bool__szPipeline.__bool__NcCs g|_dSrc)rrHr:r:r;rrzPipeline.resetcCs |dS)zClose the pipelineNrrHr:r:r;r zPipeline.closecOs|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. )rrhrmr:r:r;pipeline_execute_commands z!Pipeline.pipeline_execute_commandcOs|j|i|S)zAdds a command to the stack)rr9rnrur:r:r;rlszPipeline.execute_commandcCs<|jjs |jz|jjt|jW|S|w)z0Execute all the commands in the current pipeline)rr3rFr2Zexecute_pipelinetuplerrrHr:r:r;rws zPipeline.execute)rGrprGN)rrrrr!r<rrrintrrrrrrrlrrrwr:r:r:r;rps    rpc @seZdZdZdefddZd.ddZd/d d Zd/d d Zd/d dZ e de fddZ ddZ ddZddZddZddZddZddZ d0d!e d"efd#d$Z d0d!e d"efd%d&Z   d1d'ed(e d)eed*e dd+f d,d-ZdS)2rtz2 PubSub object for multi database client. r[cKs||_|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:)rr2rv)r9r[rur:r:r;r<szPubSub.__init__rGcCrrcr:rHr:r:r;rrzPubSub.__enter__NcCrrcrrHr:r:r;rs  zPubSub.__del__cCs|jjdS)Nrrr2Zexecute_pubsub_methodrHr:r:r;rsz PubSub.resetcCrrcrrHr:r:r;rrz PubSub.closecCs |jjjjSrc)rr2Z active_pubsub subscribedrHr:r:r;rrzPubSub.subscribedcG|jjjdg|RS)Nrlrr9rnr:r:r;rls zPubSub.execute_commandcO|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()``. psubscriberrr:r:r;r zPubSub.psubscribecGr)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrr:r:r;r zPubSub.punsubscribecOr)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()``. subscriberrr:r:r;r!rzPubSub.subscribecGr)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrr:r:r;r-szPubSub.unsubscribecOr)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()``. ssubscriberrr:r:r;r4rzPubSub.ssubscribecGr)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrr:r:r;r@rzPubSub.sunsubscribeFignore_subscribe_messagesrcC|jjjd||dS)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_messagerrrr9rrr:r:r;rI  zPubSub.get_messagecCr)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_messagerrrr:r:r;rYrzPubSub.get_sharded_message sleep_timedaemonexception_handlersharded_pubsubr cCs|jjj|||||dS)N)rrrvr)rr2Zexecute_pubsub_run)r9rrrrr:r:r; run_in_threadiszPubSub.run_in_thread)rGrtr)Fr)rFNF)rrrrr!r<rrrrpropertyrrrlrrrrrrrrrrrrr:r:r:r;rtsV         rt);loggingr4concurrent.futuresrZconcurrent.futures.threadrtypingrrrrZredis.backgroundrZ redis.backoffr Z redis.clientr Zredis.commandsr r Zredis.maint_notificationsr Zredis.multidb.circuitrrrBZredis.multidb.command_executorrZredis.multidb.configrrrrZredis.multidb.databaserrrZredis.multidb.exceptionrrrZredis.multidb.failure_detectorrZredis.multidb.healthcheckrrZredis.observability.attributesrZ redis.retryrZ redis.utilsr getLoggerrrr!rrprtr:r:r:r;s:              wC