
    phn.                     6   S SK Jr  S SKJr  S SKJr  S SKJrJrJrJ	r	J
r
Jr  S SKJr  S SKJrJr  S SKJrJrJr  S SKJrJrJrJr  S S	KJrJrJrJrJr  S S
K J!r!  S SK"J#r#  S SK$J%r%J&r&  S SK'J(r)  S SK*J+r+J,r,  S SK-J.r.  S SK/J0r0   " S S\,5      r1 " S S\+\15      r2g)    )abstractmethod)iscoroutinefunction)datetime)Any	AwaitableCallableListOptionalUnion)RedisCluster)PipelinePubSub)AsyncDatabaseDatabase	Databases)AsyncActiveDatabaseChanged&CloseConnectionOnActiveDatabaseChangedRegisterCommandFailure"ResubscribeOnActiveDatabaseChanged)DEFAULT_FAILOVER_ATTEMPTSDEFAULT_FAILOVER_DELAYAsyncFailoverStrategyDefaultFailoverStrategyExecutorFailoverStrategyExecutor)AsyncFailureDetector)Retry)AsyncOnCommandsFailEventEventDispatcherInterface)State)BaseCommandExecutorCommandExecutor)DEFAULT_AUTO_FALLBACK_INTERVAL)KeyTc                      \ rS rSr\\S\4S j5       5       r\\S\\	   4S j5       5       r
\S\	SS4S j5       r\\S\\   4S j5       5       r\S	\SS4S
 j5       r\\S\\   4S j5       5       r\R$                  \S\SS4S j5       5       r\\S\4S j5       5       r\\S\4S j5       5       r\S 5       r\S 5       r\S\4S j5       r\S\\/S4   4S j5       r\S\4S j5       r\S\ S\!4S j5       r"Sr#g)AsyncCommandExecutor   returnc                     g)zReturns a list of databases.N selfs    `/home/james-whalen/.local/lib/python3.13/site-packages/redis/asyncio/multidb/command_executor.py	databasesAsyncCommandExecutor.databases         	    c                     g)z$Returns a list of failure detectors.Nr)   r*   s    r,   failure_detectors&AsyncCommandExecutor.failure_detectors&   r/   r0   failure_detectorNc                     g)z=Adds a new failure detector to the list of failure detectors.Nr)   r+   r4   s     r,   add_failure_detector)AsyncCommandExecutor.add_failure_detector,   s     	r0   c                     g)z"Returns currently active database.Nr)   r*   s    r,   active_database$AsyncCommandExecutor.active_database1   r/   r0   databasec                    #    g7f)z#Sets the currently active database.Nr)   )r+   r<   s     r,   set_active_database(AsyncCommandExecutor.set_active_database7   
      	   c                     g)z Returns currently active pubsub.Nr)   r*   s    r,   active_pubsub"AsyncCommandExecutor.active_pubsub<   r/   r0   pubsubc                     g)zSets currently active pubsub.Nr)   r+   rE   s     r,   rC   rD   B   r/   r0   c                     g)z#Returns failover strategy executor.Nr)   r*   s    r,   failover_strategy_executor/AsyncCommandExecutor.failover_strategy_executorH   r/   r0   c                     g)zReturns command retry object.Nr)   r*   s    r,   command_retry"AsyncCommandExecutor.command_retryN   r/   r0   c                    #    g7f)z:Initializes a PubSub object on a currently active databaseNr)   r+   kwargss     r,   rE   AsyncCommandExecutor.pubsubT   r@   rA   c                    #    g7f)z*Executes a command and returns the result.Nr)   )r+   argsoptionss      r,   execute_command$AsyncCommandExecutor.execute_commandY   r@   rA   command_stackc                    #    g7f)z)Executes a stack of commands in pipeline.Nr)   )r+   rW   s     r,   execute_pipeline%AsyncCommandExecutor.execute_pipeline^   r@   rA   transactionc                    #    g7f)z1Executes a transaction block wrapped in callback.Nr)   )r+   r[   watchesrT   s       r,   execute_transaction(AsyncCommandExecutor.execute_transactionc   s
     
 	rA   method_namec                    #    g7f)z*Executes a given method on active pub/sub.Nr)   )r+   r`   rS   rP   s       r,   execute_pubsub_method*AsyncCommandExecutor.execute_pubsub_methodj   r@   rA   
sleep_timec                    #    g7f)z!Executes pub/sub run in a thread.Nr)   )r+   rd   rP   s      r,   execute_pubsub_run'AsyncCommandExecutor.execute_pubsub_runo   r@   rA   r)   )$__name__
__module____qualname____firstlineno__propertyr   r   r-   r	   r   r2   r7   r
   r   r:   r>   r   rC   setterr   rI   r   rL   rE   rU   tuplerY   r   r   r^   strrb   floatr   rf   __static_attributes__r)   r0   r,   r%   r%      s    9    4(<#=    5I d   -!8    - D   x/    F t    ,D    u        E   #XJ$45  s   5 s  r0   r%   c                     ^  \ rS rSr\\\4S\\   S\	S\
S\S\S\S\S	\4U 4S
 jjjr\S\	4S j5       r\S\\   4S j5       rS\SS4S jr\S\\   4S j5       rS\SS4S jr\S\\   4S j5       r\R4                  S\SS4S j5       r\S\4S j5       r\S\
4S j5       rS rS rS\ 4S jr!SSSS.S\"S /\#\$\%\$   4   4   S!\&S"\\'   S#\(S$\\   4
S% jjr)S&\'4S' jr* S3S(\S\$4S) jjr+ S4S*\"S+\ 4S, jjr,S- r-S. r.S/\ 4S0 jr/S1 r0S2r1U =r2$ )5DefaultCommandExecutoru   r2   r-   rL   failover_strategyevent_dispatcherfailover_attemptsfailover_delayauto_fallback_intervalc	                   > [         T
U ]  U5        U H  n	U	R                  U S9  M     X l        Xl        X0l        [        XFU5      U l        XPl        SU l	        SU l
        0 U l        U R                  5         U R                  5         g)aX  
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+   r2   r-   rL   ru   rv   rw   rx   ry   fd	__class__s             r,   r}   DefaultCommandExecutor.__init__v   s    0 	/0#B##T#: $ $"3++J.,
( "24804%'"$$&$$&r0   r'   c                     U R                   $ N)r   r*   s    r,   r-    DefaultCommandExecutor.databases   s    r0   c                     U R                   $ r   )r   r*   s    r,   r2   (DefaultCommandExecutor.failure_detectors   s    &&&r0   r4   Nc                 :    U R                   R                  U5        g r   )r   appendr6   s     r,   r7   +DefaultCommandExecutor.add_failure_detector   s    &&'78r0   c                     U R                   $ r   )r   r*   s    r,   r:   &DefaultCommandExecutor.active_database   s    $$$r0   r<   c                    #    U R                   nXl         UbI  X!LaD  U R                  R                  [        UU R                   U 40 U R                  D65      I S h  vN   g g g  N7fr   )r   r   dispatch_asyncr   r   )r+   r<   
old_actives      r,   r>   *DefaultCommandExecutor.set_active_database   st     **
 (!j&@((77*)) 00	   'A!s   AA$A"A$c                     U R                   $ r   r   r*   s    r,   rC   $DefaultCommandExecutor.active_pubsub       """r0   rE   c                     Xl         g r   r   rG   s     r,   rC   r      s    $r0   c                     U R                   $ r   )r   r*   s    r,   rI   1DefaultCommandExecutor.failover_strategy_executor   s    ///r0   c                     U R                   $ r   )r   r*   s    r,   rL   $DefaultCommandExecutor.command_retry   r   r0   c                     U R                   cf  [        U R                  R                  [        5      (       a  [        S5      eU R                  R                  R                  " S0 UD6U l         Xl        g g )Nz(PubSub is not supported for RedisClusterr)   )r   
isinstancer   clientr   
ValueErrorrE   r   rO   s     r,   rE   DefaultCommandExecutor.pubsub   s`    &$//66EE !KLL"&"7"7">">"E"E"O"OD)/& 'r0   c                 X   ^ ^^#    UUU 4S jnT R                  UT5      I S h  vN $  N7f)Nc                     >#    TR                   R                  R                  " T0 TD6I S h  vN n TR                  T5      I S h  vN   U $  N N7fr   )r   r   rU   _register_command_execution)responserS   rT   r+   s    r,   callback8DefaultCommandExecutor.execute_command.<locals>.callback   sW     !2299II  H 224888O	 9s!   *AAAAAA_execute_with_failure_detection)r+   rS   rT   r   s   ``` r,   rU   &DefaultCommandExecutor.execute_command   s&     	 99(DIIIIs   *(*rW   c                 T   ^ ^#    UU 4S jnT R                  UT5      I S h  vN $  N7f)Nc                  l  >#    TR                   R                  R                  5        IS h  vN n T H  u  pU R                  " U0 UD6  M     U R	                  5       I S h  vN nTR                  T5      I S h  vN   UsS S S 5      IS h  vN   $  Ne N2 N N! , IS h  vN  (       d  f       g = f7fr   )r   r   pipelinerU   executer   )pipecommandrT   r   rW   r+   s       r,   r   9DefaultCommandExecutor.execute_pipeline.<locals>.callback   s     ,,33<<>>$(5$G(('=W= )6 "&/66}EEE ?>> 0E ?>>>sh   )B4BB41B!B"B:B;B B4BB4BBB4B1 B#!B1-B4r   )r+   rW   r   s   `` r,   rY   'DefaultCommandExecutor.execute_pipeline   s&     	  99(MRRRRs   (&(F
shard_hintvalue_from_callablewatch_delayfuncr   r]   r   r   r   c                b   ^ ^^^^^#    UU UUUU4S jnT R                  U5      I S h  vN $  N7f)Nc                     >#    TR                   R                  R                  " T/TQ7TTTS.6I S h  vN n TR                  S5      I S h  vN   U $  N N7f)Nr   r)   )r   r   r[   r   )r   r   r+   r   r   r   r]   s    r,   r   <DefaultCommandExecutor.execute_transaction.<locals>.callback   sg     !2299EE &$7' H 222666O 7s!   0AAAAAAr   )r+   r   r   r   r   r]   r   s   `````` r,   r^   *DefaultCommandExecutor.execute_transaction   s)     		 		 99(CCCCs    /-/r`   c                 ^   ^ ^^^#    UUUU 4S jnT R                   " U/TQ76 I S h  vN $  N7f)Nc                     >#    [        TR                  T5      n [        U 5      (       a  U " T0 TD6I S h  vN nOU " T0 TD6nTR                  T5      I S h  vN   U$  N( N7fr   )getattrrC   r   r   )methodr   rS   rP   r`   r+   s     r,   r   >DefaultCommandExecutor.execute_pubsub_method.<locals>.callback  sg     T//=F"6**!'!8!88!4262224888O 9 9s!   2A"A!A"A A" A"r   )r+   r`   rS   rP   r   s   ```` r,   rb   ,DefaultCommandExecutor.execute_pubsub_method   s.     	 	 99(JTJJJJs    -+-rd   c                 Z   ^ ^^^#    UUU U4S jnT R                  U5      I S h  vN $  N7f)Nc                  V   >#    TR                   R                  TT TS9I S h  vN $  N7f)N)poll_timeoutexception_handlerrE   )r   run)r   rE   r+   rd   s   r,   r   ;DefaultCommandExecutor.execute_pubsub_run.<locals>.callback  s7     ,,00'"3 1    s   )')r   )r+   rd   r   rE   r   s   ```` r,   rf   )DefaultCommandExecutor.execute_pubsub_run  s)     	 	 99(CCCCs   +)+r   cmdsc                 ~   ^ ^^^#    UU 4S jmT R                   R                  U4S jUU 4S j5      I Sh  vN $  N7f)z?
Execute a commands execution callback with failure detection.
c                  b   >#    TR                  5       I S h  vN   T " 5       I S h  vN $  N N7fr   )_check_active_database)r   r+   s   r,   wrapperGDefaultCommandExecutor._execute_with_failure_detection.<locals>.wrapper   s+     --///!## 0#s   /+/-//c                     > T " 5       $ r   r)   )r   s   r,   <lambda>HDefaultCommandExecutor._execute_with_failure_detection.<locals>.<lambda>&  s    GIr0   c                 *   > TR                   " U /TQ76 $ r   )_on_command_fail)errorr   r+   s    r,   r   r   '  s    $//==r0   N)r   call_with_retry)r+   r   r   r   s   ```@r,   r   6DefaultCommandExecutor._execute_with_failure_detection  s6     	$
 ((88=
 
 	
 
s   0=;=c                   #    U R                   bi  U R                   R                  R                  [        R                  :w  d7  U R
                  [        :w  an  U R                  [        R                  " 5       ::  aJ  U R                  U R                  R                  5       I Sh  vN 5      I Sh  vN   U R                  5         ggg N# N7f)z2
Checks if active a database needs to be updated.
N)r   circuitstateCBStateCLOSED_auto_fallback_intervalr"   _next_fallback_attemptr   nowr>   r   r   r   r*   s    r,   r   -DefaultCommandExecutor._check_active_database*  s     
 !!)$$,,22gnnD,,0NN//8<<>A **66>>@@   ((* B O
 As$   BC	!C"C	-C.C	C	c                 h   #    U R                   R                  [        X!5      5      I S h  vN   g  N7fr   )r   r   r   )r+   r   rS   s      r,   r   'DefaultCommandExecutor._on_command_fail;  s+     $$33$T1
 	
 	
s   (202cmdc                 h   #    U R                    H  nUR                  U5      I S h  vN   M     g  N	7fr   )r   register_command_execution)r+   r   detectors      r,   r   2DefaultCommandExecutor._register_command_execution@  s,     //H55c::: 0:s   $20
2c                     [        U R                  5      n[        5       n[        5       nU R                  R                  [        U/[        UU/05        g)z 
Registers necessary listeners.
N)r   r   r   r   r   register_listenersr   r   )r+   failure_listenerresubscribe_listenerclose_connection_listeners       r,   r   .DefaultCommandExecutor._setup_event_dispatcherD  sW     2$2I2IJAC$J$L!11(+;*<*-(-	
r0   )r   r   r   r   r   r   r   r   )NN)r)   )3rh   ri   rj   rk   r   r   r"   r	   r   r   r   r   r   intrp   r}   rl   r-   r2   r7   r
   r   r:   r>   r   rC   rm   r   rI   rL   rE   rU   rn   rY   r   r   r   r   r#   ro   boolr^   rb   rf   r   r   r   r   r   rq   __classcell__)r   s   @r,   rs   rs   u   sW    "; 6(F(' 45(' (' 	('
 1(' 3(' (' (' !&(' ('T 9   '4(<#= ' '95I 9d 9 %-!8 % %- D  #x/ # # %F %t % % 0,D 0 0 #u # #0J
SE 
S  %)$)'+D
|U3	#+>%??@D D SM	D
 "D e_D*Ks K AE
D
D	
D 13
 
(-
"+"

;U ;
 
r0   rs   N)3abcr   asyncior   r   typingr   r   r   r	   r
   r   redis.asyncior   redis.asyncio.clientr   r   redis.asyncio.multidb.databaser   r   r   redis.asyncio.multidb.eventr   r   r   r   redis.asyncio.multidb.failoverr   r   r   r   r   &redis.asyncio.multidb.failure_detectorr   redis.asyncio.retryr   redis.eventr   r   redis.multidb.circuitr   r   redis.multidb.command_executorr    r!   redis.multidb.configr"   redis.typingr#   r%   rs   r)   r0   r,   <module>r     sm     '  B B & 1 M M   H % J 2 O ? S? Sl^
02F ^
r0   