o
    Ljlb                     @   s  d dl Z d dlZd dlZd dlmZmZmZmZmZ d dl	m
Z
mZ d dlmZ d dlmZ d dlmZ d dlmZmZ d dlmZ d d	lmZ d d
lmZ d dlmZ d dlmZ d dlm Z m!Z!m"Z"m#Z# d dl$m%Z%m&Z&m'Z' d dl(m)Z)m*Z*m+Z+ d dl,m-Z- d dl.m/Z/ d dl0m1Z1 d dl2m3Z3m4Z4m5Z5 d dl6m7Z7 e8e9Z:e7G dd deeZ;defddZ<G dd deeZ=G dd dZ>dS )    N)AnyCallableListLiteralOptional)HealthCheckHealthCheckPolicy)BackgroundScheduler)	NoBackoff)PubSubWorkerThread)CoreCommandsRedisModuleCommands)	DataError)MaintNotificationsConfig)CircuitBreaker)State)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheckMultiDbConfig)Database	DatabasesSyncDatabase)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)FailureDetector)GeoFailoverReason)Retry)ChannelTPubSubHandlerSubscription)experimentalc                   @   s  e Zd ZdZdefddZdd Zdd Zd	efd
dZ	de
d	dfddZ	dEdedefddZde
de
fddZdefddZde
defddZdefddZd efd!d"Zd#d$ Zd%Zd&ed'ed	efd(d)Zd&ed'ed	efd*d+Zd&ed'ed	efd,d-Zd&ed'ed	efd.d/Zd0d1 Zd2e d3gdf fd4d5Z!d6d7 Z"de
d	efd8d9Z#d	e$eef fd:d;Z%d<d= Z&d>e'd?e(d@e(fdAdBZ)dCdD Z*dS )FMultiDBClientz
    Client that operates on multiple logical Redis databases.
    Should be used in Client-side geographic failover database setups.
    configc              
   C   s   |  | _|js| n|j| _|j| _|j | _	|j
s!| n|j
| _|jd u r.| n|j| _| j| j |j| _|j| _|j| _| jtf t| j| j| j| j|j|j| j| jd| _d| _t | _t ! | _"|| _#d S )N)failure_detectors	databasescommand_retryfailover_strategyfailover_attemptsfailover_delayevent_dispatcherauto_fallback_intervalF)$r'   
_databaseshealth_checksdefault_health_checks_health_checkshealth_check_interval_health_check_intervalhealth_check_policyvalue_health_check_policyr&   default_failure_detectors_failure_detectorsr)   default_failover_strategy_failover_strategyset_databasesr-   _auto_fallback_intervalr,   _event_dispatcherr(   _command_retryupdate_supported_errorsConnectionRefusedErrorr   r*   r+   command_executorinitializedr	   _bg_scheduler	threadingLock_hc_lock_config)selfr%    rI   R/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/redis/multidb/client.py__init__,   sF   






zMultiDBClient.__init__c                 C   $   z|    W d S  ty   Y d S w N)close	ExceptionrH   rI   rI   rJ   __del__V   s
   zMultiDBClient.__del__c                 C   sv   | j | j | j | j| j d}| jD ]\}}|j| j	 |jj
tjkr/|s/|| j_d}q|s6tdd| _dS )zT
        Perform initialization of databases to define their initial state.
        FTz4Initial connection failed - no active database foundN)rC   run_coro_sync_perform_initial_health_checkrun_recurring_coror3   _check_databases_healthr.   circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDrA   _active_databaser   rB   )rH   is_active_db_founddatabaseweightrI   rI   rJ   
initialize_   s"   	
zMultiDBClient.initializereturnc                 C   s   | j S )zE
        Returns a sorted (by weight) list of all databases.
        )r.   rP   rI   rI   rJ   get_databases   s   zMultiDBClient.get_databasesr^   Nc                 C   s|   d}| j D ]\}}||krd} nq|std| j| j| |jjtjkr:| j 	dd \}}|t
jf| j_dS td)zL
        Promote one of the existing databases to become an active.
        NT/Given database is not a member of database list   r   z1Cannot set active database, database is unhealthy)r.   
ValueErrorrC   rR   _check_db_healthrV   rY   rZ   r[   	get_top_nr   MANUALrA   active_databaser   )rH   r^   existsexisting_db_highest_weighted_dbrI   rI   rJ   set_active_database   s$   z!MultiDBClient.set_active_databaseTskip_initial_health_checkc                 C   s"  t dt d|jd< d|jvrtdd|jd< |jr(| jjj|jfi |j}n"|jr@|jt dt d | jjj|jd}n
| jjdi |j}|j	du rS|
 n|j	}t|||j|jd	}z
| j| j| W n tyv   |st Y nw | jd
d \}}| j||j | || dS )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   )retriesbackoffretrymaint_notifications_configF)enabled)connection_poolN)clientrV   r_   health_check_urlrd   rI   )r   r
   client_kwargsr   from_urlrG   client_class	from_pool	set_retryrV   default_circuit_breakerr   r_   rw   rC   rR   rf   r   r.   rg   add_change_active_database)rH   r%   ro   rv   rV   r^   rm   highest_weightrI   rI   rJ   add_database   sH   

zMultiDBClient.add_databasenew_databasehighest_weight_databasec                 C   s4   |j |j kr|jjtjkr|tjf| j_d S d S d S rM   )	r_   rV   rY   rZ   r[   r   	AUTOMATICrA   ri   )rH   r   r   rI   rI   rJ   r      s   z%MultiDBClient._change_active_databasec                 C   sP   | j |}| j dd \}}||kr$|jjtjkr&|tjf| j	_
dS dS dS )z<
        Removes a database from the database list.
        rd   r   N)r.   removerg   rV   rY   rZ   r[   r   rh   rA   ri   )rH   r^   r_   rm   r   rI   rI   rJ   remove_database   s   zMultiDBClient.remove_databaser_   c                 C   sh   d}| j D ]\}}||krd} nq|std| j dd \}}| j || ||_| || dS )z<
        Updates a database from the database list.
        NTrc   rd   r   )r.   re   rg   update_weightr_   r   )rH   r^   r_   rj   rk   rl   rm   r   rI   rI   rJ   update_database_weight   s   z$MultiDBClient.update_database_weightfailure_detectorc                 C   s   | j | dS )z>
        Adds a new failure detector to the database.
        N)r8   append)rH   r   rI   rI   rJ   add_failure_detector  s   z"MultiDBClient.add_failure_detectorhealthcheckc                 C   s8   | j  | j| W d   dS 1 sw   Y  dS )z:
        Adds a new health check to the database.
        N)rF   r1   r   )rH   r   rI   rI   rJ   add_health_check  s   "zMultiDBClient.add_health_checkc                 O   s    | j s|   | jj|i |S )zB
        Executes a single command and return its result.
        )rB   r`   rA   execute_commandrH   argsoptionsrI   rI   rJ   r     s   zMultiDBClient.execute_commandzEHIMPORT is not supported on the multi-database (Active-Active) clientr   kwargsc                 O   
   t | jrM   r   _HIMPORT_UNSUPPORTEDrH   r   r   rI   rI   rJ   himport_prepare/     
zMultiDBClient.himport_preparec                 O   r   rM   r   r   rI   rI   rJ   himport_set2  r   zMultiDBClient.himport_setc                 O   r   rM   r   r   rI   rI   rJ   himport_discard5  r   zMultiDBClient.himport_discardc                 O   r   rM   r   r   rI   rI   rJ   himport_discard_all8  r   z!MultiDBClient.himport_discard_allc                 C   s   t | S )z:
        Enters into pipeline mode of the client.
        )PipelinerP   rI   rI   rJ   pipeline;  s   zMultiDBClient.pipelinefuncr   c                 O   s&   | j s|   | jj|g||R  S )z3
        Executes callable as transaction.
        )rB   r`   rA   execute_transaction)rH   r   watchesr   rI   rI   rJ   transactionA  s   zMultiDBClient.transactionc                 K   s   | j s|   t| fi |S )z
        Return a Publish/Subscribe object. With this object, you can
        subscribe to channels and listen for messages that get published to
        them.
        )rB   r`   PubSub)rH   r   rI   rI   rJ   pubsubJ  s   zMultiDBClient.pubsubc                    s   | j  t| j}W d   n1 sw   Y  | j||I dH }|s3|jjtjkr1tj|j_|S |rA|jjtj	krAtj	|j_|S )zO
        Runs health checks on the given database until first failure.
        N)
rF   listr1   r6   executerV   rY   rZ   OPENr[   )rH   r^   r/   
is_healthyrI   rI   rJ   rf   U  s   

zMultiDBClient._check_db_healthc                    s   i  g | _ | jD ]\}}t| |}| |< | j | q	tj| j ddiI dH } fddt| j |D }| D ]\}}t	|t
rZ|j}tj|j_tjd|jd d||< q=|S )	zk
        Runs health checks as a recurring task.
        Runs health checks against all databases.
        return_exceptionsTNc                    s   i | ]	\}} | |qS rI   rI   ).0taskresult
task_to_dbrI   rJ   
<dictcomp>x  s    z9MultiDBClient._check_databases_health.<locals>.<dictcomp>z%Health check failed, due to exception)exc_infoF)	_hc_tasksr.   asynciocreate_taskrf   r   gatherzipitems
isinstancer   r^   rZ   r   rV   rY   loggerdebugoriginal_exception)rH   r^   rl   r   results
db_resultsr   unhealthy_dbrI   r   rJ   rU   h  s,   



z%MultiDBClient._check_databases_healthc                    s   |   I dH }d}| jjtjkrd| v}n!| jjtjkr,t| t|d k}n| jjtj	kr9d| v }|sDt
d| jj dS )zj
        Runs initial health check and evaluate healthiness based on initial_health_check_policy.
        NTF   z:Initial health check failed. Initial health check policy: )rU   rG   initial_health_check_policyr   ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumlenONE_AVAILABLEr   )rH   r   r   rI   rI   rJ   rS     s    z+MultiDBClient._perform_initial_health_checkrV   	old_state	new_statec                 C   s   |t jkr| j| j|j d S |t jkr,|t jkr,t	d|j d | j
tt| |t jkrB|t jkrDtd|j d d S d S d S )Nz	Database z- is unreachable. Failover has been initiated.z is reachable again.)rZ   	HALF_OPENrC   run_coro_fire_and_forgetrf   r^   r[   r   r   warningrun_oncer   _half_open_circuitinfo)rH   rV   r   r   rI   rI   rJ   rX     s   
z/MultiDBClient._on_circuit_state_change_callbackc                 C   sV   | j rz
| j | jj W n	 ty   Y nw | j   | jjr)| jjj  dS dS )z:
        Closes the client and all its resources.
        N)	rC   rR   r6   rN   rO   stoprA   ri   rv   rP   rI   rI   rJ   rN     s   
zMultiDBClient.close)T)+__name__
__module____qualname____doc__r   rK   rQ   r`   r   rb   r   rn   r   boolr   r   r   r   floatr   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   rf   dictrU   rS   r   rZ   rX   rN   rI   rI   rI   rJ   r$   %   sV    *	&
8
	"
r$   rV   c                 C   s   t j| _d S rM   )rZ   r   rY   )rV   rI   rI   rJ   r        r   c                   @   s   e Zd ZU dZdZed ed< defddZd'dd	Z	d
d Z
dd ZdefddZdefddZd(ddZd(ddZd'ddZdd ZdededefddZdededefdd Zdededefd!d"Zdededefd#d$Zdee fd%d&ZdS ))r   zG
    Pipeline implementation for multiple logical Redis databases.
    F_is_async_clientrv   c                 C   s   g | _ || _d S rM   )_command_stack_client)rH   rv   rI   rI   rJ   rK     s   
zPipeline.__init__ra   c                 C      | S rM   rI   rP   rI   rI   rJ   	__enter__     zPipeline.__enter__c                 C      |    d S rM   reset)rH   exc_type	exc_value	tracebackrI   rI   rJ   __exit__  r   zPipeline.__exit__c                 C   rL   rM   r   rO   rP   rI   rI   rJ   rQ     s
   zPipeline.__del__c                 C   s
   t | jS rM   )r   r   rP   rI   rI   rJ   __len__  r   zPipeline.__len__c                 C   s   dS )z1Pipeline instances should always evaluate to TrueTrI   rP   rI   rI   rJ   __bool__  s   zPipeline.__bool__Nc                 C   s
   g | _ d S rM   )r   rP   rI   rI   rJ   r     r   zPipeline.resetc                 C   s   |    dS )zClose the pipelineNr   rP   rI   rI   rJ   rN        zPipeline.closec                 O   s   | 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.
        )r   r   r   rI   rI   rJ   pipeline_execute_command  s   z!Pipeline.pipeline_execute_commandc                 O   s   | j |i |S )zAdds a command to the stack)r   r   rI   rI   rJ   r     s   zPipeline.execute_commandr   r   c                 O   
   t tjrM   r   r$   r   r   rI   rI   rJ   r   
  r   zPipeline.himport_preparec                 O   r   rM   r   r   rI   rI   rJ   r     r   zPipeline.himport_setc                 O   r   rM   r   r   rI   rI   rJ   r     r   zPipeline.himport_discardc                 O   r   rM   r   r   rI   rI   rJ   r     r   zPipeline.himport_discard_allc                 C   s<   | j js	| j   z| j jt| jW |   S |   w )z0Execute all the commands in the current pipeline)r   rB   r`   rA   execute_pipelinetupler   r   rP   rI   rI   rJ   r     s   
zPipeline.execute)ra   r   ra   N)r   r   r   r   r   r   __annotations__r$   rK   r   r   rQ   intr   r   r   r   rN   r   r   r   r   r   r   r   r   r   rI   rI   rI   rJ   r     s$   
 



r   c                   @   s  e Zd ZdZdefddZd0ddZd1d	d
Zd1ddZd1ddZ	e
defddZdd ZdeeB deddfddZdd ZdeeB deddfddZdd ZdeeB deddfddZdd  Z	"d2d#ed$efd%d&Z	"d2d#ed$efd'd(Z	"	!		!d3d)ed*ed+ee d,edd-f
d.d/ZdS )4r   z2
    PubSub object for multi database client.
    rv   c                 K   s   || _ | j jjdi | dS )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
        NrI   )r   rA   r   )rH   rv   r   rI   rI   rJ   rK   (  s   zPubSub.__init__ra   c                 C   r   rM   rI   rP   rI   rI   rJ   r   3  r   zPubSub.__enter__Nc                 C   rL   rM   r   rP   rI   rI   rJ   rQ   6  s
   zPubSub.__del__c                 C   s   | j jdS )Nr   r   rA   execute_pubsub_methodrP   rI   rI   rJ   r   ?  s   zPubSub.resetc                 C   r   rM   r   rP   rI   rI   rJ   rN   B  r   zPubSub.closec                 C   s   | j jjjS rM   )r   rA   active_pubsub
subscribedrP   rI   rI   rJ   r   E  r   zPubSub.subscribedc                 G      | j jjdg|R  S )Nr   r   rH   r   rI   rI   rJ   r   I  s
   zPubSub.execute_commandr   r   c                 O      | j jjdg|R 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()``.
        
psubscriber   r   rI   rI   rJ   r   N     
zPubSub.psubscribec                 G   r   )zj
        Unsubscribe from the supplied patterns. If empty, unsubscribe from
        all patterns.
        punsubscriber   r   rI   rI   rJ   r   \  
   zPubSub.punsubscribec                 O   r   )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()``.
        	subscriber   r   rI   rI   rJ   r   e  r   zPubSub.subscribec                 G   r   )zi
        Unsubscribe from the supplied channels. If empty, unsubscribe from
        all channels
        unsubscriber   r   rI   rI   rJ   r   s  s   zPubSub.unsubscribec                 O   r   )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()``.
        
ssubscriber   r   rI   rI   rJ   r   z  r   zPubSub.ssubscribec                 G   r   )zu
        Unsubscribe from the supplied shard_channels. If empty, unsubscribe from
        all shard_channels
        sunsubscriber   r   rI   rI   rJ   r     r   zPubSub.sunsubscribeF        ignore_subscribe_messagestimeoutc                 C      | j jjd||d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_messager   r   r   rH   r   r   rI   rI   rJ   r    
   
zPubSub.get_messagec                 C   r  )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_messager  r   r  rI   rI   rJ   r    r  zPubSub.get_sharded_message
sleep_timedaemonexception_handlersharded_pubsubr   c                 C   s   | j jj|||| |dS )N)r  r	  r   r
  )r   rA   execute_pubsub_run)rH   r  r  r	  r
  rI   rI   rJ   run_in_thread  s   zPubSub.run_in_thread)ra   r   r   )Fr   )r   FNF)r   r   r   r   r$   rK   r   rQ   r   rN   propertyr   r   r   r    r"   r!   r   r   r   r   r   r   r   r  r  r   r   r  rI   rI   rI   rJ   r   #  sz    


	

	




r   )?r   loggingrD   typingr   r   r   r   r   !redis.asyncio.multidb.healthcheckr   r   redis.backgroundr	   redis.backoffr
   redis.clientr   redis.commandsr   r   redis.exceptionsr   redis.maint_notificationsr   redis.multidb.circuitr   r   rZ   redis.multidb.command_executorr   redis.multidb.configr   r   r   r   redis.multidb.databaser   r   r   redis.multidb.exceptionr   r   r   redis.multidb.failure_detectorr   redis.observability.attributesr   redis.retryr   redis.typingr    r!   r"   redis.utilsr#   	getLoggerr   r   r$   r   r   r   rI   rI   rI   rJ   <module>   s>    
   'U