,iHjddlZddlZddlmZddlmZddlmZmZm Z m Z ddl m Z ddl mZddlmZmZddlmZdd lmZdd lmZdd lmZmZdd lmZmZmZdd lm Z m!Z!ddl"m#Z#ddl$m%Z%m&Z&ddl'm(Z(ejRe*Z+e(GddeeZ,defdZ-GddeeZ.GddZ/y)N) as_completed)ThreadPoolExecutor)AnyCallableListOptional)BackgroundScheduler)PubSubWorkerThread) CoreCommandsRedisModuleCommands)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIOD MultiDbConfig)Database Databases SyncDatabase)NoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector) HealthCheckHealthCheckPolicy) experimentalceZdZdZdefdZdZdefdZde ddfd Z de fd Z d e d e fd Z de fdZde defdZdefdZdefdZdZdZdedgdffdZdZde defdZd#deegdffdZdeded efd!Z d"Z!y)$ MultiDBClientzx Client that operates on multiple logical Redis databases. Should be used in Active-Active database setups. configc |j|_|js|jn |j|_|j |_|jj|j|j|_ |js|jn |j|_|j|j!n |j|_|j"j%|j|j&|_|j*|_|j.|_|j0j3t4ft7|j|j|j0|j"|j8|j:|j,|j(|_d|_tAjB|_"tG|_$||_%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_policyvaluehealth_check_probeshealth_check_probes_delay_health_check_policyrdefault_failure_detectors_failure_detectorsr"default_failover_strategy_failover_strategy set_databasesr&_auto_fallback_intervalr%_event_dispatcherr!_command_retryupdate_supported_errorsConnectionRefusedErrorrr#r$command_executor initialized threadingRLock_hc_lockr _bg_scheduler_config)selfrs Y/var/www/html/langgraph-service/venv/lib/python3.12/site-packages/redis/multidb/client.py__init__zMultiDBClient.__init__s **,''  ( ( *%%  '-&B&B#7=7Q7Q7W7W  & &(H(H8 ! ++  , , .)) ''/  , , .))  --doo>'-'D'D$!'!8!8$22 335K4MN 6"55oo--"55$66!00!33#'#?#? ! !!) 02 cd}|j||jj|j|jd}|jD]h\}}|j j |j|j jtjk(sS|rV||j_ d}j|s tdd|_y)zT Perform initialization of databases to define their initial state. c|N)errors rDraise_exception_on_failed_hcz>MultiDBClient.initialize..raise_exception_on_failed_hcMsKrF)on_errorFTz4Initial connection failed - no active database foundN)_check_databases_healthrA run_recurringr,r'circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDr<active_databaserr=)rCrLis_active_db_founddatabaseweights rD initializezMultiDBClient.initializeHs   $$.J$K ((  ' '  ( ( # $ * Hf    - -d.T.T U%%7@R8@%%5%)" *"*F  rFreturnc|jS)zE Returns a sorted (by weight) list of all databases. )r'rCs rD get_databaseszMultiDBClient.get_databasesksrFrXNcFd}|jD]\}}||k(s d}n|s td|j||jjt j k(r3|jjdd\}}||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' ValueError_check_db_healthrPrSrTrU get_top_nr<rVr)rCrXexists existing_db_highest_weighted_dbs rDset_active_databasez!MultiDBClient.set_active_databaseqs"oo NKh&  NO O h'    ! !W^^ 3%)__%>%>q%A!%D " 4 Adds a new failure detector to the database. N)r3append)rCrxs rDadd_failure_detectorz"MultiDBClient.add_failure_detectors &&'78rF healthcheckc||j5|jj|dddy#1swYyxYw)z: Adds a new health check to the database. N)r@r*rz)rCr|s rDadd_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=rZr<execute_commandrCargsoptionss rDrzMultiDBClient.execute_commands5 OO 4t$$44dFgFFrFct|S)z: Enters into pipeline mode of the client. )Pipeliner]s rDpipelinezMultiDBClient.pipelines~rFfuncrcx|js|j|jj|g||S)z3 Executes callable as transaction. )r=rZr<execute_transaction)rCrwatchesrs rD transactionzMultiDBClient.transactions: OO 8t$$88RR'RRrFc 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=rZPubSub)rCkwargss rDpubsubzMultiDBClient.pubsubs'  OO d%f%%rFcv|jj|j|}|sH|jjt j k7rt j |j_|S|rF|jjt jk7rt j|j_|S)zO Runs health checks on the given database until first failure. )r1executer*rPrSrTOPENrU)rCrX is_healthys rDrczMultiDBClient._check_db_healths ..66t7J7JHU %%5)0  &  H,,22gnnD%,^^H   "rFrMc btt|j5}|jDchc]!\}}|j|j|#}}} t ||j D]} |j dddycc}}w#t$rj}|j}tj|j_ tjd|j |r||j Yd}~d}~wwxYw#t"$r t#dwxYw#1swYyxYw)zk Runs health checks as a recurring task. Runs health checks against all databases. ) max_workers)timeoutz%Health check failed, due to exception)exc_infoNz4Health check execution exceeds health_check_interval)rlenr'submitrcrr,resultrrXrTrrPrSlogger exceptionoriginal_exception TimeoutError) rCrMexecutorrXrgfuturesfuturee unhealthy_dbs rDrNz%MultiDBClient._check_databases_healths   C,@ A X$(??Ha 5 5x@G  *T%@%@;F ;  ;  6 ;'(zz 5<\\ ,,2((C%&%9%9) $$Q%9%9: ;  "J 1  sYD%&BD%D 5BD D% D  A DD D  D  D""D%%D.rP old_state new_statec|tjk(r|j|jy|tjk(r:|tj k(r&|j jtt|yyyrI) rT HALF_OPENrcrXrUrrArun_oncer_half_open_circuit)rCrPrrs rDrRz/MultiDBClient._on_circuit_state_change_callback#se )) )  ! !'"2"2 3   &9 +D    ' '$&8' ,E &rFc|jr|jj|jjr/|jjjj yy)z: Closes the client and all its resources. N)rAstopr<rVclientcloser]s rDrzMultiDBClient.close/sQ       # # %  0 0  ! ! 1 1 8 8 > > @ 1rFrI)"__name__ __module__ __qualname____doc__rrErZrr^rrirnrlrrtfloatrwrr{rr~rrrrrboolrc ExceptionrNr rTrRrrJrFrDrrs (}(T! Fy  L T 2 D\ DA(ACOA H HD|DUD&9_9 4K4G S*t); <S &$ )d9J0K D % 29 FM ArFrrPc.tj|_yrI)rTrrS)rPs rDrr9s%%GMrFcxeZdZdZdefdZddZdZdZde fdZ de fd Z dd Z dd Zdd ZdZdeefdZy )rzG Pipeline implementation for multiple logical Redis databases. rc g|_||_yrI)_command_stack_client)rCrs rDrEzPipeline.__init__Bs  rFr[c|SrIrJr]s rD __enter__zPipeline.__enter__F rFc$|jyrIreset)rCexc_type exc_value tracebacks rD__exit__zPipeline.__exit__I  rFcD |jy#t$rYywxYwrIrrr]s rD__del__zPipeline.__del__Ls"  JJL     c,t|jSrI)rrr]s rD__len__zPipeline.__len__Rs4&&''rFcy)z1Pipeline instances should always evaluate to TrueTrJr]s rD__bool__zPipeline.__bool__UsrFNcg|_yrI)rr]s rDrzPipeline.resetYs  rFc$|jy)zClose the pipelineNrr]s rDrzPipeline.close\s  rFc@|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. )rrzrs rDpipeline_execute_commandz!Pipeline.pipeline_execute_command`s! ""D'?3 rFc&|j|i|S)zAdds a command to the stack)rrCrrs rDrzPipeline.execute_commandos,t,,d=f==rFc|jjs|jj |jjj t |j |jS#|jwxYw)z0Execute all the commands in the current pipeline)rr=rZr<execute_pipelinetuplerrr]s rDrzPipeline.executess_||'' LL # # % <<00AAd))* JJLDJJLs 7A::B )r[rr[N)rrrrrrErrrintrrrrrrrrrrrJrFrDrr=s^} (($! > c rFrc 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. rc ^||_|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 NrJ)rr<r)rCrrs rDrEzPubSub.__init__s( , %%,,6v6rFr[c|SrIrJr]s rDrzPubSub.__enter__rrFNcD |jy#t$rYywxYwrIrr]s rDrzPubSub.__del__s$  JJL   rcL|jjjdS)Nrrr<execute_pubsub_methodr]s rDrz PubSub.resets||,,BB7KKrFc$|jyrIrr]s rDrz PubSub.closerrFcV|jjjjSrI)rr< active_pubsub subscribedr]s rDrzPubSub.subscribeds||,,::EEErFcP|jjjdg|S)NrrrCrs rDrzPubSub.execute_commands,Bt||,,BB  $  rFcV|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 rDrzPubSub.psubscribe7Ct||,,BB   #)  rFcP|jjjdg|S)zj Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. punsubscriberrs rDrzPubSub.punsubscribe/ Ct||,,BB  !  rFcV|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 rDrzPubSub.subscribes7Ct||,,BB   "(  rFcP|jjjdg|S)zi Unsubscribe from the supplied channels. If empty, unsubscribe from all channels unsubscriberrs rDrzPubSub.unsubscribes( Ct||,,BB=XSWXXrFcV|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 rDrzPubSub.ssubscriberrFcP|jjjdg|S)zu Unsubscribe from the supplied shard_channels. If empty, unsubscribe from all shard_channels sunsubscriberrs rDrzPubSub.sunsubscriberrFignore_subscribe_messagesrcR|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_messagerrrrCrrs rDrzPubSub.get_messages0||,,BB &?C  rFcR|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 rDrzPubSub.get_sharded_messages0||,,BB !&?C  rF sleep_timedaemonexception_handlersharded_pubsubr cV|jjj|||||S)N)rrrr)rr<execute_pubsub_run)rCrrrrs rD run_in_threadzPubSub.run_in_threads6||,,?? /) @  rF)r[rr)F)rFNF)rrrrrrErrrrpropertyrrrrrrrrrrrrrrrrJrFrDrrs 7} 7LFDFF     Y   IL )- @E "IL )- @E $ 04$      $H-      rFr)0loggingr>concurrent.futuresrconcurrent.futures.threadrtypingrrrrredis.backgroundr redis.clientr redis.commandsr r redis.multidb.circuitr rrTredis.multidb.command_executorrredis.multidb.configrrredis.multidb.databaserrrredis.multidb.exceptionrrredis.multidb.failure_detectorrredis.multidb.healthcheckrr redis.utilsr getLoggerrrrrrrrJrFrDrs+8000+<02ADDDX:D$   8 $^A'^A^AB &&@"L@FU U rF