o z$AsyncCommandExecutor.execute_command command_stackcr5)z)Executes a stack of commands in pipeline.Nr()r*rBr(r(r+execute_pipelinegr>z%AsyncCommandExecutor.execute_pipeline transactioncr5)z1Executes a transaction block wrapped in callback.Nr()r*rDwatchesr@r(r(r+execute_transactionlsz(AsyncCommandExecutor.execute_transaction method_namecr5)z*Executes a given method on active pub/sub.Nr()r*rGr?r=r(r(r+execute_pubsub_methodsr>z*AsyncCommandExecutor.execute_pubsub_method sleep_timecr5)z!Executes pub/sub run in a thread.Nr()r*rIr=r(r(r+execute_pubsub_runxr>z'AsyncCommandExecutor.execute_pubsub_run)#__name__ __module__ __qualname__propertyrrr,rrr.r1r rr2r#r6r r7setterrr:rr;r8rAtuplerCrr rFstrrHfloatrrJr(r(r(r+r%!s\      r%cseZdZeeefdeedede de de de de de ffd d Zed efd d Zed eefddZded dfddZed eefddZdeded dfddZed eefddZejded dfddZed efddZed e fd d!Zd"d#Zd$d%Zd&e fd'd(Z!dd)dd*d+e"d,ge#e$e%e$ffd-e&d.ee'd/e(d0ee f d1d2Z)d3e'fd4d5Z* dGd6e d e$fd7d8Z+ 9dHd:e"d;e fdd?Z-d@dAZ.dBe fdCdDZ/dEdFZ0Z1S)IDefaultCommandExecutorr.r,r;failover_strategyevent_dispatcherfailover_attemptsfailover_delayauto_fallback_intervalc snt||D]} | j|dq||_||_||_t||||_||_d|_ d|_ i|_ | | dS)a Initialize the DefaultCommandExecutor instance. Args: failure_detectors: List of failure detector instances to monitor database health databases: Collection of available databases to execute commands on command_retry: Retry policy for failed command execution failover_strategy: Strategy for handling database failover event_dispatcher: Interface for dispatching events failover_attempts: Number of failover attempts failover_delay: Delay between failover attempts auto_fallback_interval: Time interval in seconds between attempts to fall back to a primary database )command_executorN)super__init__set_command_executor _databases_failure_detectors_command_retryr_failover_strategy_executor_event_dispatcher_active_database_active_pubsub_active_pubsub_kwargs_setup_event_dispatcher_schedule_next_fallback) r*r.r,r;rTrUrVrWrXfd __class__r(r+r[s  zDefaultCommandExecutor.__init__r&cC|jSN)r]r)r(r(r+r,z DefaultCommandExecutor.databasescCrjrk)r^r)r(r(r+r.rlz(DefaultCommandExecutor.failure_detectorsr/NcCs|j|dSrk)r^appendr0r(r(r+r1sz+DefaultCommandExecutor.add_failure_detectorcCrjrk)rbr)r(r(r+r2rlz&DefaultCommandExecutor.active_databaser3r4csd|j}||_|dur.||ur0t|||dIdH|jt||j|fi|jIdHdSdSdS)N) fail_fromfail_tor4)rbrradispatch_asyncrrd)r*r3r4 old_activer(r(r+r6s& z*DefaultCommandExecutor.set_active_databasecCrjrkrcr)r(r(r+r7rlz$DefaultCommandExecutor.active_pubsubr8cCs ||_dSrkrrr9r(r(r+r7s cCrjrk)r`r)r(r(r+r:rlz1DefaultCommandExecutor.failover_strategy_executorcCrjrk)r_r)r(r(r+r;rlz$DefaultCommandExecutor.command_retrycKsD|jdur t|jjtrtd|jjjdi||_||_dSdS)Nz(PubSub is not supported for RedisClusterr()rc isinstancerbclientr ValueErrorr8rdr<r(r(r+r8s  zDefaultCommandExecutor.pubsubcs$fdd}|IdHS)Ncs0jjjiIdH}IdH|Srk)rbrtrA_register_command_executionresponser?r@r*r(r+callbacks z8DefaultCommandExecutor.execute_command..callback_execute_with_failure_detection)r*r?r@rzr(ryr+rAsz&DefaultCommandExecutor.execute_commandrBcs"fdd}|IdHS)Nc sjj4IdH+}D] \}}|j|i|q|IdH}IdH|WdIdHS1IdHs.callbackr{)r*rBrzr(rr+rCs z'DefaultCommandExecutor.execute_pipelineF shard_hintvalue_from_callable watch_delayfuncr rErrrcs(fdd}|IdHS)Ncs<jjjgRdIdH}dIdH|S)Nrr()rbrtrDrvrwrr*rrrrEr(r+rzsz.callbackr{)r*rrrrrErzr(rr+rFs z*DefaultCommandExecutor.execute_transactionrGcs,fdd}j|gRIdHS)NcsNtj}t|r|iIdH}n|i}IdH|Srk)getattrr7rrv)methodrxr?r=rGr*r(r+rzs z>DefaultCommandExecutor.execute_pubsub_method..callbackr{)r*rGr?r=rzr(rr+rHs z,DefaultCommandExecutor.execute_pubsub_methodrIcs$fdd}|IdHS)NcsjjdIdHS)N) poll_timeoutexception_handlerr8)rcrunr(rr8r*rIr(r+rz s  z;DefaultCommandExecutor.execute_pubsub_run..callbackr{)r*rIrr8rzr(rr+rJsz)DefaultCommandExecutor.execute_pubsub_runr(rzcmdscs6fddjfddfddIdHS)zO Execute a commands execution callback with failure detection. csIdHIdHSrk)_check_active_databaser()rzr*r(r+wrapper0s zGDefaultCommandExecutor._execute_with_failure_detection..wrappercsSrkr(r()rr(r+6szHDefaultCommandExecutor._execute_with_failure_detection..csj|gRSrk)_on_command_fail)error)rr*r(r+r7sN)r_call_with_retry)r*rzrr()rzrr*rr+r|)s    z6DefaultCommandExecutor._execute_with_failure_detectioncsh|jdus|jjjtjks|jdkr0|jtkr2| |j IdHt j IdH|dSdSdS)zB Checks if active a database needs to be updated. Nr)rbcircuitstateCBStateCLOSED_auto_fallback_interval_next_fallback_attemptrnowr6r`r~r# AUTOMATICrfr)r(r(r+r:s   z-DefaultCommandExecutor._check_active_databasecs|jt||IdHdSrk)rarpr)r*rr?r(r(r+rLsz'DefaultCommandExecutor._on_command_failcmdcs"|jD] }||IdHqdSrk)r^register_command_execution)r*rdetectorr(r(r+rvQs z2DefaultCommandExecutor._register_command_executioncCs4t|j}t}t}|jt|gt||gidS)z0 Registers necessary listeners. N)rr^rrraregister_listenersrr)r*failure_listenerresubscribe_listenerclose_connection_listenerr(r(r+reUs z.DefaultCommandExecutor._setup_event_dispatcher)NN)r()2rKrLrMrrr"rrrrrrintrRr[rNr,r.r1r rr2r#r6r r7rOrr:r;r8rArPrCrr rrr$rQboolrFrHrJr|rrrvre __classcell__r(r(rhr+rS~s *      rSN)7abcrasynciorrtypingrrrrr r redis.asyncior redis.asyncio.clientr r redis.asyncio.multidb.databaserrrredis.asyncio.multidb.eventrrrrredis.asyncio.multidb.failoverrrrrr&redis.asyncio.multidb.failure_detectorr$redis.asyncio.observability.recorderrredis.asyncio.retryr redis.eventrrredis.multidb.circuitrrredis.multidb.command_executorr r!redis.multidb.configr"redis.observability.attributesr# redis.typingr$r%rSr(r(r(r+s(           ]