o
    Lj:\                     @   sz  d dl Z d dlZd dlmZmZmZmZmZmZ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 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*m+Z+m,Z, d dl-m.Z. d dl/m0Z0m1Z1m2Z2m3Z3m4Z4 d dl5m6Z6 e7e8Z9e6G 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	AwaitableCallableListLiteralOptionalUnion)DefaultCommandExecutor)DEFAULT_GRACE_PERIODDatabaseConfigInitialHealthCheckMultiDbConfig)AsyncDatabaseDatabase	Databases)AsyncFailureDetector)HealthCheckHealthCheckPolicy)Retry)BackgroundScheduler)	NoBackoff)AsyncCoreCommandsAsyncRedisModuleCommands)	DataError)CircuitBreaker)State)InitialHealthCheckFailedErrorNoValidDatabaseExceptionUnhealthyDatabaseException)GeoFailoverReason)ChannelT
EncodableTKeyTPubSubHandlerSubscription)experimentalc                   @   s  e Zd ZdZdefddZdNddZd	d
 Zdd Zdd Z	de
fddZdeddfddZ	dOd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/d0Zd+ed,edefd1d2Zd+ed,edefd3d4Zd5d6 Z dd7dd8d9e!d:ge"ee#e f f d;e$d<e%e& d=ed>e%e f
d?d@Z'dAdB Z(de)e*ef fdCdDZ+dEdF Z,dedefdGdHZ-dIe.dJe/dKe/fdLdMZ0dS )P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g t| j| j| j| j|j|j| j| jd| _d| _t | _ t! | _"|| _#d | _$g | _%d | _&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initializedasyncioLock_hc_lockr   _bg_scheduler_config_recurring_hc_task	_hc_tasks_half_open_state_task)selfr'    rN   Z/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/redis/asyncio/multidb/client.py__init__)   sL   






zMultiDBClient.__init__rM   returnc                    s   | j s|  I d H  | S N)rD   
initializerM   rN   rN   rO   
__aenter__V   s   zMultiDBClient.__aenter__c                    sj   | j r	| j   | jr| j  | jD ]}|  q| j I d H  | jjr3| jjj	 I d H  d S d S rR   )
rJ   cancelrL   rK   r8   closerC   active_databaseclientaclose)rM   hc_taskrN   rN   rO   rZ   [   s   



zMultiDBClient.aclosec                       |   I d H  d S rR   rZ   rM   exc_type	exc_value	tracebackrN   rN   rO   	__aexit__k      zMultiDBClient.__aexit__c                    s   |   I dH  t| j| j| j| _d}| jD ]\}}|j	
| j |j	jtjkr4|s4|| j_d}q|s;tdd| _dS )zT
        Perform initialization of databases to define their initial state.
        NFTz4Initial connection failed - no active database found)_perform_initial_health_checkrE   create_taskrH   run_recurring_asyncr5   _check_databases_healthrJ   r0   circuiton_state_changed!_on_circuit_state_change_callbackstateCBStateCLOSEDrC   _active_databaser   rD   )rM   is_active_db_founddatabaseweightrN   rN   rO   rS   n   s(   
zMultiDBClient.initializec                 C   s   | j S )zE
        Returns a sorted (by weight) list of all databases.
        )r0   rT   rN   rN   rO   get_databases   s   zMultiDBClient.get_databasesrp   Nc                    s   d}| j D ]\}}||krd} nq|std| |I dH  |jjtjkr?| j dd \}}| j	|t
jI dH  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)r0   
ValueError_check_db_healthrh   rk   rl   rm   	get_top_nrC   set_active_databaser   MANUALr   )rM   rp   existsexisting_db_highest_weighted_dbrN   rN   rO   rx      s&   
z!MultiDBClient.set_active_databaseTskip_initial_health_checkc                    s  |j dtdt di |jr| jjj|jfi |j }n"|jr7|jtdt d | jjj|jd}n
| jjdi |j }|j	du rJ|
 n|j	}t|||j|jd}z
| |I dH  W n tym   |sk Y nw | jdd \}}| j||j | ||I dH  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.
        retryr   )retriesbackoff)connection_poolN)rY   rh   rq   health_check_urlrt   rN   )client_kwargsupdater   r   from_urlrI   client_class	from_pool	set_retryrh   default_circuit_breakerr   rq   r   rv   r   r0   rw   add_change_active_database)rM   r'   r~   rY   rh   rp   r}   highest_weightrN   rN   rO   add_database   sD   
zMultiDBClient.add_databasenew_databasehighest_weight_databasec                    s>   |j |j kr|jjtjkr| j|tjI d H  d S d S d S rR   )	rq   rh   rk   rl   rm   rC   rx   r   	AUTOMATIC)rM   r   r   rN   rN   rO   r      s   z%MultiDBClient._change_active_databasec                    sZ   | j |}| j dd \}}||kr)|jjtjkr+| j|t	j
I dH  dS dS dS )z<
        Removes a database from the database list.
        rt   r   N)r0   removerw   rh   rk   rl   rm   rC   rx   r   ry   )rM   rp   rq   r}   r   rN   rN   rO   remove_database   s   zMultiDBClient.remove_databaserq   c                    sp   d}| j D ]\}}||krd} nq|std| j dd \}}| j || ||_| ||I dH  dS )z<
        Updates a database from the database list.
        NTrs   rt   r   )r0   ru   rw   update_weightrq   r   )rM   rp   rq   rz   r{   r|   r}   r   rN   rN   rO   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)r:   append)rM   r   rN   rN   rO   add_failure_detector  s   z"MultiDBClient.add_failure_detectorhealthcheckc              	      sN   | j 4 I dH  | j| W d  I dH  dS 1 I dH s w   Y  dS )z:
        Adds a new health check to the database.
        N)rG   r3   r   )rM   r   rN   rN   rO   add_health_check  s   .zMultiDBClient.add_health_checkc                    s.   | j s|  I dH  | jj|i |I dH S )zB
        Executes a single command and return its result.
        N)rD   rS   rC   execute_commandrM   argsoptionsrN   rN   rO   r     s   zMultiDBClient.execute_commandzEHIMPORT is not supported on the multi-database (Active-Active) clientr   kwargsc                       t | jrR   r   _HIMPORT_UNSUPPORTEDrM   r   r   rN   rN   rO   himport_prepare2     
zMultiDBClient.himport_preparec                    r   rR   r   r   rN   rN   rO   himport_set5  r   zMultiDBClient.himport_setc                    r   rR   r   r   rN   rN   rO   himport_discard8  r   zMultiDBClient.himport_discardc                    r   rR   r   r   rN   rN   rO   himport_discard_all;  r   z!MultiDBClient.himport_discard_allc                 C   s   t | S )z:
        Enters into pipeline mode of the client.
        )PipelinerT   rN   rN   rO   pipeline>  s   zMultiDBClient.pipelineF
shard_hintvalue_from_callablewatch_delayfuncr   watchesr   r   r   c                   s:   | j s|  I dH  | jj|g|R |||dI dH S )z3
        Executes callable as transaction.
        Nr   )rD   rS   rC   execute_transaction)rM   r   r   r   r   r   rN   rN   rO   transactionD  s   zMultiDBClient.transactionc                    s&   | j s|  I dH  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.
        N)rD   rS   PubSub)rM   r   rN   rN   rO   pubsubZ  s   zMultiDBClient.pubsubc                    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 rN   rN   ).0taskresult
task_to_dbrN   rO   
<dictcomp>u  s    z9MultiDBClient._check_databases_health.<locals>.<dictcomp>z%Health check failed, due to exception)exc_infoF)rK   r0   rE   re   rv   r   gatherzipitems
isinstancer   rp   rl   OPENrh   rk   loggerdebugoriginal_exception)rM   rp   r|   r   results
db_resultsr   unhealthy_dbrN   r   rO   rg   e  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: )rg   rI   initial_health_check_policyr   ALL_AVAILABLEvaluesMAJORITY_AVAILABLEsumlenONE_AVAILABLEr   )rM   r   
is_healthyrN   rN   rO   rd     s    z+MultiDBClient._perform_initial_health_checkc                    sX   | j | j|I dH }|s|jjtjkrtj|j_|S |r*|jjtjkr*tj|j_|S )zO
        Runs health checks on the given database until first failure.
        N)r8   executer3   rh   rk   rl   r   rm   )rM   rp   r   rN   rN   rO   rv     s   


zMultiDBClient._check_db_healthrh   	old_state	new_statec                 C   s   t  }|tjkrt | |j| _d S |tjkr0|tj	kr0t
d|j d |tt| |tjkrF|tjkrHt
d|j d d S d S d S )Nz	Database z- is unreachable. Failover has been initiated.z is reachable again.)rE   get_running_looprl   	HALF_OPENre   rv   rp   rL   rm   r   r   warning
call_laterr
   _half_open_circuitinfo)rM   rh   r   r   looprN   rN   rO   rj     s   

z/MultiDBClient._on_circuit_state_change_callback)rM   r&   rQ   r&   )T)1__name__
__module____qualname____doc__r   rP   rU   rZ   rb   rS   r   rr   r   rx   r   boolr   r   r   floatr   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r"   r   strr   r   dictr   rg   rd   rv   r   rl   rj   rN   rN   rN   rO   r&   "   sr    
-$
1


"r&   rh   c                 C   s   t j| _d S rR   )rl   r   rk   )rh   rN   rN   rO   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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.
    T_is_async_clientrY   c                 C   s   g | _ || _d S rR   )_command_stack_client)rM   rY   rN   rN   rO   rP     s   
zPipeline.__init__rM   rQ   c                       | S rR   rN   rT   rN   rN   rO   rU        zPipeline.__aenter__c                    s*   |   I d H  | j|||I d H  d S rR   )resetr   rb   r^   rN   rN   rO   rb     s   zPipeline.__aexit__c                 C   s   |    S rR   )_async_self	__await__rT   rN   rN   rO   r     r   zPipeline.__await__c                    r   rR   rN   rT   rN   rN   rO   r     r   zPipeline._async_selfc                 C   s
   t | jS rR   )r   r   rT   rN   rN   rO   __len__     
zPipeline.__len__c                 C   s   dS )z1Pipeline instances should always evaluate to TrueTrN   rT   rN   rN   rO   __bool__  s   zPipeline.__bool__Nc                    s   g | _ d S rR   )r   rT   rN   rN   rO   r     r   zPipeline.resetc                    s   |   I dH  dS )zClose the pipelineN)r   rT   rN   rN   rO   rZ     s   zPipeline.aclosec                 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   rN   rN   rO   pipeline_execute_command  s   z!Pipeline.pipeline_execute_commandc                 O   s   | j |i |S )zAdds a command to the stack)r   r   rN   rN   rO   r     s   zPipeline.execute_commandr   r   c                 O   
   t tjrR   r   r&   r   r   rN   rN   rO   r     r   zPipeline.himport_preparec                 O   r   rR   r   r   rN   rN   rO   r   
  r   zPipeline.himport_setc                 O   r   rR   r   r   rN   rN   rO   r     r   zPipeline.himport_discardc                 O   r   rR   r   r   rN   rN   rO   r     r   zPipeline.himport_discard_allc                    sV   | j js| j  I dH  z| j jt| jI dH W |  I dH  S |  I dH  w )z0Execute all the commands in the current pipelineN)r   rD   rS   rC   execute_pipelinetupler   r   rT   rN   rN   rO   r     s   
 zPipeline.execute)rM   r   rQ   r   rQ   N)rQ   r   )r   r   r   r   r   r   __annotations__r&   rP   rU   rb   r   r   intr   r   r   r   rZ   r   r   r   r   r   r   r   r   r   rN   rN   rN   rO   r     s&   
 



r   c                   @   s   e Zd ZdZdefddZd&ddZd'd	d
Zdd Ze	de
fddZdefddZdeeB deddfddZdefddZdeeB deddfddZdd Z	d(de
dee fdd Zdd!d"d#eddfd$d%ZdS ))r   z2
    PubSub object for multi database client.
    rY   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
        NrN   )r   rC   r   )rM   rY   r   rN   rN   rO   rP   %  s   zPubSub.__init__rQ   c                    r   rR   rN   rT   rN   rN   rO   rU   0  r   zPubSub.__aenter__Nc                    r\   rR   r]   r^   rN   rN   rO   rb   3  rc   zPubSub.__aexit__c                    s   | j jdI d H S )NrZ   r   rC   execute_pubsub_methodrT   rN   rN   rO   rZ   6  s   zPubSub.aclosec                 C   s   | j jjjS rR   )r   rC   active_pubsub
subscribedrT   rN   rN   rO   r   9  s   zPubSub.subscribedr   c                    s   | j jjdg|R  I d H S )Nr   r   rM   r   rN   rN   rO   r   =  s   zPubSub.execute_commandr   c                    $   | j jjdg|R i |I dH 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()``.
        
psubscribeNr   r   rN   rN   rO   r   B     
zPubSub.psubscribec                       | j jjdg|R  I dH S )zj
        Unsubscribe from the supplied patterns. If empty, unsubscribe from
        all patterns.
        punsubscribeNr   r   rN   rN   rO   r   P     zPubSub.punsubscribec                    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()``.
        	subscribeNr   r   rN   rN   rO   r  Y  r   zPubSub.subscribec                    r   )zi
        Unsubscribe from the supplied channels. If empty, unsubscribe from
        all channels
        unsubscribeNr   r   rN   rN   rO   r  g  r  zPubSub.unsubscribeF        ignore_subscribe_messagestimeoutc                    s   | j jjd||dI dH 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  Nr   )rM   r  r  rN   rN   rO   r  p  s   
zPubSub.get_messageg      ?)exception_handlerpoll_timeoutr	  c                   s   | j jj||| dI dH S )a  Process pub/sub messages using registered callbacks.

        This is the equivalent of :py:meth:`redis.PubSub.run_in_thread` in
        redis-py, but it is a coroutine. To launch it as a separate task, use
        ``asyncio.create_task``:

            >>> task = asyncio.create_task(pubsub.run())

        To shut it down, use asyncio cancellation:

            >>> task.cancel()
            >>> await task
        )
sleep_timer  r   N)r   rC   execute_pubsub_run)rM   r  r	  rN   rN   rO   run  s   z
PubSub.run)rQ   r   r   )Fr  )r   r   r   r   r&   rP   rU   rb   rZ   propertyr   r   r!   r   r    r$   r#   r   r   r  r  r   r   r  r  rN   rN   rN   rO   r      sL    


	


r   )>rE   loggingtypingr   r   r   r   r   r   r   &redis.asyncio.multidb.command_executorr	   redis.asyncio.multidb.configr
   r   r   r   redis.asyncio.multidb.databaser   r   r   &redis.asyncio.multidb.failure_detectorr   !redis.asyncio.multidb.healthcheckr   r   redis.asyncio.retryr   redis.backgroundr   redis.backoffr   redis.commandsr   r   redis.exceptionsr   redis.multidb.circuitr   r   rl   redis.multidb.exceptionr   r   r   redis.observability.attributesr   redis.typingr    r!   r"   r#   r$   redis.utilsr%   	getLoggerr   r   r&   r   r   r   rN   rN   rN   rO   <module>   s8    $
   &V