File: //home/arjun/projects/env/lib64/python3.10/site-packages/redis/__pycache__/cluster.cpython-310.pyc
o
$weRj �
@ s� d dl Z d dlZd dlZd dlZd dlZd dlmZ d dlmZm Z m
Z
m Z m Z m
Z
mZ d dlmZmZ d dlmZ d dlmZ d dlmZmZmZ d dlmZm Z d d l m Z d d
l m Z m!Z!m"Z" d d l#m$Z$m%Z% d d l&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5 d d
l6m7Z7 d dl8m9Z9 d dl:m;Z;m<Z<m=Z=m>Z>m?Z?m@Z@ deAdeeAeBf deAfdd�ZCdd� ZDdd� ZEdd� ZFded ede
e
eBeBf e
eAef f fd d �ZGd d � ZHd!d"� ZId#ZJd$ZKd%ZLd&ZMd'ZNd(d)� ZOG d*d+� d+e!�ZPG d,d-� d-�ZQG d.d/� d/eQe �ZRG d0d1� d1�ZSG d2d3� d3�ZTG d4d5� d5�ZUG d6d7� d7e�ZVG d8d9� d9eR�ZWd:eAde d;ef fd<d=�ZXd>ZYeYD ]ZZeZ�[d?d@��\� ZZe]eWeZeXeZ�� �q>G dAdB� dB�Z^G dCdD� dD�Z_dS )E� N)� OrderedDict)�Any�Callable�Dict�List�Optional�Tuple�Union)�CommandsParser�Encoder)�
parse_scan)�default_backoff)�CaseInsensitiveDict�PubSub�Redis)�
READ_COMMANDS�RedisClusterCommands)� list_or_args)�ConnectionPool�
DefaultParser� parse_url)�REDIS_CLUSTER_HASH_SLOTS�key_slot)�AskError�AuthenticationError�ClusterCrossSlotError�ClusterDownError� ClusterError�ConnectionError� DataError�MasterDownError�
MovedError�RedisClusterException�
RedisError�
ResponseError�SlotNotCoveredError� TimeoutError�
TryAgainError��Lock)�Retry)�HIREDIS_AVAILABLE�
dict_merge�list_keys_to_dict� merge_result�safe_str� str_if_bytes�host�port�returnc C s | � d|� �S )N�:� �r1 r2 r5 r5 �F/home/arjun/projects/env/lib/python3.10/site-packages/redis/cluster.py�
get_node_name. � r8 c O s | j p| jj|d fi |��S )Nr )�
connection�connection_pool�get_connection)�
redis_node�args�optionsr5 r5 r7 r< 2 s
��r< c K sF i }g }|� � D ]\}}t|fi |��\}}|||< ||7 }q||fS �N)�itemsr ) �command�resr? �cursors�ret� node_name�response�cursor�rr5 r5 r7 �parse_scan_result8 s
rJ c
K sf t � }|�� D ] }|D ]\}}z
|| |7 < W q ty& |||< Y q w qdd� |�� D �}|S )Nc S s g | ]\}}||f�qS r5 r5 )�.0�channel�numsubr5 r5 r7 �
<listcomp>L � z'parse_pubsub_numsub.<locals>.<listcomp>)r �values�KeyErrorrA )rB rC r? �numsub_d� numsub_tupsrL � numsubbed�
ret_numsubr5 r5 r7 �parse_pubsub_numsubC s ��rV �respr? c s~ |� dd�� dtdtttf f� fdd� �i }| D ]#}|d d� \}}}|dd � }�|� �fdd �|D �d
�|||f< q|S ) N� current_host� r> r3 c s t | d �p� | d fS �Nr � )r0 �r> )rX r5 r7 �
fix_serverU s z'parse_cluster_slots.<locals>.fix_server� c s g | ]}� |� �qS r5 r5 )rK �replica)r] r5 r7 rN ^ � z'parse_cluster_slots.<locals>.<listcomp>)�primary�replicas)�getr r �str)rW r? �slots�slot�start�endra rb r5 )rX r] r7 �parse_cluster_slotsP s �ri c K s� t | d t�r | S g }| D ]S}g g d�}tdt|d �d�D ]}|d �|d | |d |d f� q |d }|D ] }i }tdt|�d�D ] }||d ||| < qF|d �|� q:|�|� q
|S )z(
Parse CLUSTER SHARDS response.
r )re �nodesr[ � re r^ rj )�
isinstance�dict�range�len�append) rW r? �shards�x�shard�irj �node� dict_noder5 r5 r7 �parse_cluster_shardsd s
( rw c K s
| � d�S )z+
Parse CLUSTER MYSHARDID response.
�utf-8)�decode)rW r? r5 r5 r7 �parse_cluster_myshardidz s
rz ra r_ zslot-id)$�charset�connection_classr; �connection_pool_class� client_name�credential_provider�db�decode_responses�encoding�encoding_errors�errorsr1 �lib_name� lib_version�max_connections�
nodes_flag�redis_connect_func�passwordr2 � queue_class�retry�retry_on_timeout�protocol�socket_connect_timeout�socket_keepalive�socket_keepalive_options�socket_timeout�ssl� ssl_ca_certs� ssl_ca_data� ssl_certfile�
ssl_cert_reqs� ssl_keyfile� ssl_password�unix_socket_path�usernamer6 c K s dd� | � � D �}|S )z9
Remove unsupported or disabled keys from kwargs
c S s& i | ]\}}|t v r|tvr||�qS r5 )�REDIS_ALLOWED_KEYS�KWARGS_DISABLED_KEYS)rK �k�vr5 r5 r7 �
<dictcomp>� s
�z"cleanup_kwargs.<locals>.<dictcomp>)rA )�kwargs�connection_kwargsr5 r5 r7 �cleanup_kwargs� s �r� c @ s& e Zd Zeejeeee e
e d��ZdS )�
ClusterParser)�ASK�TRYAGAIN�MOVED� CLUSTERDOWN� CROSSSLOT�
MASTERDOWNN) �__name__�
__module__� __qualname__r, r �EXCEPTION_CLASSESr r' r! r r r r5 r5 r5 r7 r� � s ��r� c @ s e Zd ZdZdZdZdZdZdZeeeeehZ e
e g d�e�e g d�e�e d ge�e g d
�e ��Z
g d �fZeeed �Ze
e d
dge�e dgdd� �e g d�e�e g d�dd� �e ddgdd� �e dgdd� �e dge�e dgd d� �e d gd d� �e d gd d� ��
ZeeefZd'd(d%d&�Zd!S ))�AbstractRedisCluster� � primariesrb �all�randomz default-node)WzACL CATz ACL DELUSERz
ACL DRYRUNz ACL GENPASSz ACL GETUSERzACL HELPzACL LISTzACL LOGzACL LOADzACL SAVEz ACL SETUSERz ACL USERSz
ACL WHOAMI�AUTH� CLIENT LISTzCLIENT SETINFO�CLIENT SETNAME�CLIENT GETNAME�
CONFIG SET�CONFIG REWRITE�CONFIG RESETSTAT�TIME�PUBSUB CHANNELS�
PUBSUB NUMPAT�
PUBSUB NUMSUB�PUBSUB SHARDCHANNELS�PUBSUB SHARDNUMSUB�PING�INFO�SHUTDOWN�KEYS�DBSIZE�BGSAVE� SLOWLOG GET� SLOWLOG LEN�
SLOWLOG RESET�WAIT�WAITAOF�SAVE� MEMORY PURGEzMEMORY MALLOC-STATSz MEMORY STATS�LASTSAVEzCLIENT TRACKINGINFO� CLIENT PAUSE�CLIENT UNPAUSE�CLIENT UNBLOCKz CLIENT IDz CLIENT REPLYzCLIENT GETREDIRz CLIENT INFO� CLIENT KILL�READONLYz CLUSTER INFOz CLUSTER MEET�CLUSTER MYSHARDIDz
CLUSTER NODESzCLUSTER REPLICASz
CLUSTER RESETzCLUSTER SET-CONFIG-EPOCH�
CLUSTER SLOTS�CLUSTER SHARDSz CLUSTER COUNT-FAILURE-REPORTSzCLUSTER KEYSLOT�COMMANDz
COMMAND COUNTz COMMAND LISTzCOMMAND GETKEYS�
CONFIG GET�DEBUG� RANDOMKEYr� � READWRITEr� zTFUNCTION LOADzTFUNCTION DELETEzTFUNCTION LIST�TFCALL� TFCALLASYNCz GRAPH.CONFIGzLATENCY HISTORYzLATENCY LATESTz
LATENCY RESETz MODULE LISTz MODULE LOADz
MODULE UNLOADz
MODULE LOADEX) �FLUSHALL�FLUSHDBzFUNCTION DELETEzFUNCTION FLUSHz
FUNCTION LISTz
FUNCTION LOADzFUNCTION RESTOREzREDISGEARS_2.REFRESHCLUSTER�SCAN�
SCRIPT EXISTS� SCRIPT FLUSH� SCRIPT LOADz
FUNCTION DUMP)zCLUSTER COUNTKEYSINSLOTzCLUSTER DELSLOTSzCLUSTER DELSLOTSRANGEzCLUSTER GETKEYSINSLOTzCLUSTER SETSLOT) z FT.CREATEz FT.SEARCHz FT.AGGREGATEz
FT.EXPLAINz
FT.EXPLAINCLIz
FT,PROFILEzFT.ALTERz FT.DROPINDEXz FT.ALIASADDzFT.ALIASUPDATEz FT.ALIASDELz
FT.TAGVALSz FT.SUGADDz FT.SUGGETz FT.SUGDELz FT.SUGLENz FT.SYNUPDATEz
FT.SYNDUMPz
FT.SPELLCHECKz
FT.DICTADDz
FT.DICTDELz FT.DICTDUMPzFT.INFOzFT._LISTz FT.CONFIGzFT.ADDzFT.DELzFT.DROPzFT.GETzFT.MGETz FT.SYNADD)r� r� r� r� r� r� c C s t t|�� ��S r@ )�sum�listrP �rB rC r5 r5 r7 �<lambda>| � z AbstractRedisCluster.<lambda>)r� r� r� ) r� r� r� r� r� r� r� r� r� r� r� c C � t |t�r t|�� �S |S r@ )rl rm r� rP r� r5 r5 r7 r� � � r� r� c C r� r@ )rl rm r� rP r� r5 r5 r7 r� � r� r� c C s t |�� �dkr
dS dS rZ )r� rP r� r5 r5 r7 r� � rO r� r� c C s t |�� ��� S r@ )r� rP �popr� r5 r5 r7 r� � r� r� c C s dd� t |�� � D �S )Nc S s g | ]}t |��qS r5 )r� )rK r� r5 r5 r7 rN � r` z1AbstractRedisCluster.<lambda>.<locals>.<listcomp>)�ziprP r� r5 r5 r7 r� � � r� c C s t |�� �S r@ )r� rP r� r5 r5 r7 r� � s N� target_node� ClusterNoder3 c st |r|| j _dS | �� � � fdd�| �� D �}|r"t�|�| j _dS � fdd�| �� D �}|r8t�|�| j _dS dS )a Replace the default cluster node.
A random cluster node will be chosen if target_node isn't passed, and primaries
will be prioritized. The default node will not be changed if there are no other
nodes in the cluster.
Args:
target_node (ClusterNode, optional): Target node to replace the default
node. Defaults to None.
c � g | ]}|� kr|�qS r5 r5 �rK ru �� curr_noder5 r7 rN � rO z=AbstractRedisCluster.replace_default_node.<locals>.<listcomp>c r� r5 r5 r� r� r5 r7 rN � rO N)�
nodes_manager� default_node�get_default_node�
get_primariesr� �choice� get_replicas)�selfr� r� rb r5 r� r7 �replace_default_node� s
�z)AbstractRedisCluster.replace_default_noder@ )r� r� r3 N) r� r� r� �RedisClusterRequestTTL� PRIMARIES�REPLICAS� ALL_NODES�RANDOM� DEFAULT_NODE�
NODE_FLAGSr, r- �SLOT_ID�
COMMAND_FLAGS�SEARCH_COMMANDSri rw rz �#CLUSTER_COMMANDS_RESPONSE_CALLBACKSrV r. rJ �RESULT_CALLBACKSr r& r �ERRORS_ALLOW_RETRYr� r5 r5 r5 r7 r� � sv Y�\�
��|�%�
��
��
�
�
��
)r� c @ s� e Zd Zedd� �Z d`d ee d
ed eed d
eded de dede de dee dee
eege eef f fdd�Z dd� Z
dd� Zd d � Zd d � Zd d!� Zd"d#� Zdad$d%�Zd&d'� Zd(d)� Zd*d+� Zd,d-� Zdbd.d/�Zd0d1� Zd2d3� Zd4ed fd5d6�Zdcd7d8�Z ddd9d:�Z dad;d<�Z ded=d>�Z ? dfd@dA�Z dBdC� Z!d4ed fdDdE�Z"dFdG� Z#dHdI� Z$dJdK� Z%dLdM� Z&dNdO� Z'dPdQ� Z(dRdS� Z)dTdU� Z*dVdW� Z+dXdY� Z,dZd[� Z-d\d]� Z.d^d_� Z/dS )g� RedisClusterc K s | dd|i|��S )a
Return a Redis client object configured from the given URL
For example::
redis://[[username]:[password]]@localhost:6379/0
rediss://[[username]:[password]]@localhost:6379/0
unix://[username@]/path/to/socket.sock?db=0[&password=password]
Three URL schemes are supported:
- `redis://` creates a TCP socket connection. See more at:
<https://www.iana.org/assignments/uri-schemes/prov/redis>
- `rediss://` creates a SSL wrapped TCP socket connection. See more at:
<https://www.iana.org/assignments/uri-schemes/prov/rediss>
- ``unix://``: creates a Unix Domain Socket connection.
The username, password, hostname, path and all querystring values
are passed through urllib.parse.unquote in order to replace any
percent-encoded values with their corresponding characters.
There are several ways to specify a database number. The first value
found will be used:
1. A ``db`` querystring option, e.g. redis://localhost?db=0
2. If using the redis:// or rediss:// schemes, the path argument
of the url, e.g. redis://localhost/0
3. A ``db`` keyword argument to this function.
If none of these options are specified, the default db=0 is used.
All querystring options are cast to their appropriate Python types.
Boolean arguments can be specified with string values "True"/"False"
or "Yes"/"No". Values that cannot be properly cast cause a
``ValueError`` to be raised. Once parsed, the querystring arguments
and keyword arguments are passed to the ``ConnectionPool``'s
class initializer. In the case of conflicting arguments, querystring
arguments always win.
�urlNr5 r5 )�clsr
r� r5 r5 r7 �from_url� s *zRedisCluster.from_urlN�� r^ F� Tr1 r2 �
startup_nodesr� � cluster_error_retry_attemptsr� r* �require_full_coverage�reinitialize_steps�read_from_replicas�dynamic_startup_nodesr
�
address_remapc K s� |du rg }d| v rt d��d}
|
durId}
t|
�}d|v r"t d��d|v r0|d dkr0t d ��| �|� | �d
�}| �d |�}|�t||�� n|durZ|durZ|�t||�� n
t|�dkrdt d ��| �d
d�| _| �d
| j i� t
di | ��} |r�|| _ | �d| j i� n | �dt t
� d�i� t| �dd�| �dd�| �dd��| _|| _| jj�� | _| jj�� | _|| _d| _|| _td||
|| | d�| ��| _t | jj �| _ t | jj �| _ t!| �| _"t#�$� | _%dS )a�
Initialize a new RedisCluster client.
:param startup_nodes:
List of nodes from which initial bootstrapping can be done
:param host:
Can be used to point to a startup node
:param port:
Can be used to point to a startup node
:param require_full_coverage:
When set to False (default value): the client will not require a
full coverage of the slots. However, if not all slots are covered,
and at least one node has 'cluster-require-full-coverage' set to
'yes,' the server will throw a ClusterDownError for some key-based
commands. See -
https://redis.io/topics/cluster-tutorial#redis-cluster-configuration-parameters
When set to True: all slots must be covered to construct the
cluster client. If not all slots are covered, RedisClusterException
will be thrown.
:param read_from_replicas:
Enable read from replicas in READONLY mode. You can read possibly
stale data.
When set to true, read commands will be assigned between the
primary and its replications in a Round-Robin manner.
:param dynamic_startup_nodes:
Set the RedisCluster's startup nodes to all of the discovered nodes.
If true (default value), the cluster's discovered nodes will be used to
determine the cluster nodes-slots mapping in the next topology refresh.
It will remove the initial passed startup nodes if their endpoints aren't
listed in the CLUSTER SLOTS output.
If you use dynamic DNS endpoints for startup nodes but CLUSTER SLOTS lists
specific IP addresses, it is best to set it to false.
:param cluster_error_retry_attempts:
Number of times to retry before raising an error when
:class:`~.TimeoutError` or :class:`~.ConnectionError` or
:class:`~.ClusterDownError` are encountered
:param reinitialize_steps:
Specifies the number of MOVED errors that need to occur before
reinitializing the whole cluster topology. If a MOVED error occurs
and the cluster does not need to be reinitialized on this current
error handling, only the MOVED slot will be patched with the
redirected node.
To reinitialize the cluster on every MOVED error, set
reinitialize_steps to 1.
To avoid reinitializing the cluster on moved errors, set
reinitialize_steps to 0.
:param address_remap:
An optional callable which, when provided with an internal network
address of a node, e.g. a `(host, port)` tuple, will return the address
where the node is reachable. This can be used to map the addresses at
which the nodes _think_ they are, to addresses at which a client may
reach them, such as when they sit behind a proxy.
:**kwargs:
Extra arguments that will be sent into Redis instance when created
(See Official redis-py doc for supported kwargs
[https://github.com/andymccurdy/redis-py/blob/master/redis/client.py])
Some kwargs are not supported and will raise a
RedisClusterException:
- db (Redis do not support database SELECT in cluster mode)
Nr� z4Argument 'db' is not possible to use in cluster modeFT�pathzFRedisCluster does not currently support Unix Domain Socket connectionsr z9A ``db`` querystring option can only be 0 in cluster moder1 r2 a5 RedisCluster requires at least one node to discover the cluster. Please provide one of the followings:
1. host and port, for example:
RedisCluster(host='localhost', port=6379)
2. list of startup nodes, for example:
RedisCluster(startup_nodes=[ClusterNode('localhost', 6379), ClusterNode('localhost', 6378)])r� r� r� rx r� �strictr� )r r r r r r5 )&r" r �updaterc rp r� ro r� �user_on_connect_func�
on_connectr� r� r* r
r �encoderr � __class__r �copy�
command_flagsr �
node_flagsr �reinitialize_counterr � NodesManagerr� r r �cluster_response_callbacksr �result_callbacksr
�commands_parser� threadingr) �_lock)r� r1 r2 r r r� r r r r r
r r� r � url_optionsr5 r5 r7 �__init__� sz L���
�
��� �
zRedisCluster.__init__c C � | S r@ r5 �r� r5 r5 r7 � __enter__� s zRedisCluster.__enter__c C � | � � d S r@ ��close�r� �exc_type� exc_value� tracebackr5 r5 r7 �__exit__� � zRedisCluster.__exit__c C r, r@ r- r* r5 r5 r7 �__del__� r4 zRedisCluster.__del__c C s< | � � D ]}|jrz|jj�� W q ty Y qw qd S r@ )� get_nodes�redis_connectionr; �
disconnect�OSError�r� ru r5 r5 r7 �disconnect_connection_pools� s ���z(RedisCluster.disconnect_connection_poolsc C sV |� t� |�� | jr |�d� t|�� �dkr td��| jdur)| �|� dS dS )z�
Initialize the connection, authenticate and select a database and send
READONLY if it is set during object initialization.
r� �OKzREADONLY command failedN) �
set_parserr� r r � send_commandr0 �
read_responser r )r� r: r5 r5 r7 r � s
�zRedisCluster.on_connectc C s\ |j s+| j� |j s| j�|g� W d � |j S W d � |j S 1 s&w Y |j S r@ )r7 r& r� �create_redis_connectionsr: r5 r5 r7 �get_redis_connection� s
��
��z!RedisCluster.get_redis_connectionc C s | j �|||�S r@ )r� �get_node�r� r1 r2 rF r5 r5 r7 rB � � zRedisCluster.get_nodec C � | j �t�S r@ )r� �get_nodes_by_server_type�PRIMARYr* r5 r5 r7 r� � r4 zRedisCluster.get_primariesc C rE r@ )r� rF �REPLICAr* r5 r5 r7 r� � r4 zRedisCluster.get_replicasc C s t �t| jj�� ��S r@ )r� r� r� r� � nodes_cacherP r* r5 r5 r7 �get_random_node� � z RedisCluster.get_random_nodec C s t | jj�� �S r@ )r� r� rI rP r* r5 r5 r7 r6 � rD zRedisCluster.get_nodesc C st | � |�}| jj�|�}|du st|�dkr td|� d���|r,t| jj| �dk r,dS |r4d}|| S d}|| S )z�
Get the node that holds the key's slot.
If replica set to True but the slot doesn't have any replicas, None is
returned.
Nr �Slot "z " is not covered by the cluster.rk r[ )�keyslotr� � slots_cacherc ro r% )r� �keyr_ rf �
slot_cache�node_idxr5 r5 r7 �get_node_from_key� s
�z RedisCluster.get_node_from_keyc C � | j jS )z0
Get the cluster's default node
)r� r� r* r5 r5 r7 r� � � z RedisCluster.get_default_nodec C s* |du s
| j |jd�du rdS || j_dS )z�
Set the default node of the cluster.
:param node: 'ClusterNode'
:return True if the default node was set, else False
N�rF FT)rB �namer� r� r: r5 r5 r7 �set_default_node� s z RedisCluster.set_default_noder3 c C � | j S r@ )r� r* r5 r5 r7 � get_retry� s zRedisCluster.get_retryc C s$ || _ | �� D ]}|j�|� qd S r@ )r� r6 r7 � set_retry)r� r� ru r5 r5 r7 rZ � s �zRedisCluster.set_retryc C s6 |du r| � � }|jdu rtd|j� d���|j�� S )ai
Returns a Monitor object for the specified target node.
The default cluster node will be selected if no target node was
specified.
Monitor is useful for handling the MONITOR command to the redis server.
next_command() method returns one command from monitor
listen() method yields commands from monitor.
Nz
Cluster Node z has no redis_connection)r� r7 r" rV �monitor)r� r� r5 r5 r7 r[ � s
�
zRedisCluster.monitorc K s t | f|||d�|��S )z~
Allows passing a ClusterNode, or host&port, to get a pubsub instance
connected to the specified node
)ru r1 r2 )�
ClusterPubSub)r� ru r1 r2 r� r5 r5 r7 �pubsub s zRedisCluster.pubsubc C sF |rt d��|r t d��t| j| j| jj| j| j| j| j| j | j
d� S )ac
Cluster impl:
Pipelines do not work in cluster mode the same way they
do in normal mode. Create a clone of this object so
that simulating pipelines will work correctly. Each
command will be called directly when used and
when calling execute() will only return the result stack.
z(shard_hint is deprecated in cluster modez)transaction is deprecated in cluster mode) r� r$ r r# r" r r r �lock) r" �ClusterPipeliner� r$ r r# r" r r r r& )r� � transaction�
shard_hintr5 r5 r7 �pipeline
s �zRedisCluster.pipeline皙�����?c C s" |du rt }|| ||||||d�S )aL
Return a new Lock object using key ``name`` that mimics
the behavior of threading.Lock.
If specified, ``timeout`` indicates a maximum life for the lock.
By default, it will remain locked until release() is called.
``sleep`` indicates the amount of time to sleep per loop iteration
when the lock is in blocking mode and another client is currently
holding the lock.
``blocking`` indicates whether calling ``acquire`` should block until
the lock has been acquired or to fail immediately, causing ``acquire``
to return False and the lock not being acquired. Defaults to True.
Note this value can be overridden by passing a ``blocking``
argument to ``acquire``.
``blocking_timeout`` indicates the maximum amount of time in seconds to
spend trying to acquire the lock. A value of ``None`` indicates
continue trying forever. ``blocking_timeout`` can be specified as a
float or integer, both representing the number of seconds to wait.
``lock_class`` forces the specified lock implementation. Note that as
of redis-py 3.0, the only lock class we implement is ``Lock`` (which is
a Lua-based lock). So, it's unlikely you'll need this parameter, unless
you have created your own custom lock class.
``thread_local`` indicates whether the lock token is placed in
thread-local storage. By default, the token is placed in thread local
storage so that a thread only sees its token, not a token set by
another thread. Consider the following timeline:
time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
thread-1 sets the token to "abc"
time: 1, thread-2 blocks trying to acquire `my-lock` using the
Lock instance.
time: 5, thread-1 has not yet completed. redis expires the lock
key.
time: 5, thread-2 acquired `my-lock` now that it's available.
thread-2 sets the token to "xyz"
time: 6, thread-1 finishes its work and calls release(). if the
token is *not* stored in thread local storage, then
thread-1 would see the token value as "xyz" and would be
able to successfully release the thread-2's lock.
In some use cases it's necessary to disable thread local storage. For
example, if you have code where one thread acquires a lock and passes
that lock instance to a worker thread to release later. If thread
local storage isn't disabled in this case, the worker thread won't see
the token set by the thread that acquired the lock. Our assumption
is that these cases aren't common and as such default to using
thread local storage.N)�timeout�sleep�blocking�blocking_timeout� thread_localr( )r� rV rd re rf rg �
lock_classrh r5 r5 r7 r^ % s >�zRedisCluster.lockc C s || j |< dS )z Set a custom Response CallbackN)r" )r� rB �callbackr5 r5 r7 �set_response_callbacko s z"RedisCluster.set_response_callbackc O s" |d � � }t|�dkr)|d � d|d � �� � | jv r)|d � d|d � �� � }|�dd �}|d ur6|}n| j�|�}|| jjkrG| �� gS || jjkrQ| � � S || jj
kr[| � � S || jj kre| �
� S || jjkrp| jjgS || jjd v r}| jjgS | j|� }| j�|| jo�|tv �}|gS )Nr rk � r[ r� )�upperro r r� rc r r rJ r� r� r� r� r� r6 r r� r� r �determine_slot�get_node_from_slotr r )r� r> r� rB r� � command_flagrf ru r5 r5 r7 �_determine_nodess s0 ,
�z RedisCluster._determine_nodesc C s | j dkrdS | j| j dkS )Nr F)r r r* r5 r5 r7 �_should_reinitialized� s
z"RedisCluster._should_reinitializedc C s | j �|�}t|�S )z�
Calculate keyslot for a given key.
See Keys distribution model in https://redis.io/topics/cluster-spec
)r �encoder )r� rO r� r5 r5 r7 rM � s zRedisCluster.keyslotc G s | � � j}| jj|g|�R � S )a�
Get the keys in the command. If the command has no keys in in, None is
returned.
NOTE: Due to a bug in redis<7.0, this function does not work properly
for EVAL or EVALSHA when the `numkeys` arg is 0.
- issue: https://github.com/redis/redis/issues/9493
- fix: https://github.com/redis/redis/pull/9733
So, don't use this function with EVAL or EVALSHA.
)r� r7 r$ �get_keys)r� r> �
redis_connr5 r5 r7 �_get_command_keys� s
z RedisCluster._get_command_keysc s |d }� j �|�tkr|d S |�� dv r@t|�dkr#td|� ���t|d �}|dd| � }t|�dkr=t�dt �S |}n"� j
|� }|du sOt|�dkrb|�� dv r[t�dt �S td |� ���t|�dkro� � |d �S � fd
d �|D �}t|�dkr�t|� d ���|� � S )
a
Figure out what slot to use based on args.
Raises a RedisClusterException if there's a missing key and we can't
determine what slots to map the command to; or, if the keys don't
all map to the same key slot.
r r[ )�EVAL�EVALSHArk zInvalid args in command: r^ N)�FCALL�FCALL_ROz�No way to dispatch this command to Redis Cluster. Missing key.
You can execute the command by specifying target nodes.
Command: c s h | ]}� � |��qS r5 )rM )rK rO r* r5 r7 � <setcomp>� r� z.RedisCluster.determine_slot.<locals>.<setcomp>z) - all keys must map to the same key slot)
r rc r rm ro r" �intr� � randranger rv rM r� )r� r> rB �num_actual_keys� eval_keys�keysre r5 r* r7 rn � s8
�� �zRedisCluster.determine_slotc C rX )z.
Get the connections' encoder
)r r* r5 r5 r7 � get_encoder� � zRedisCluster.get_encoderc C rS )z9
Get the connections' key-word arguments
)r� r� r* r5 r5 r7 �get_connection_kwargs� rT z"RedisCluster.get_connection_kwargsc C s t |t�o || jv S r@ )rl rd r )r� � target_nodesr5 r5 r7 �_is_nodes_flag s zRedisCluster._is_nodes_flagc C sN t |t�r |}|S t |t�r|g}|S t |t�r |�� }|S tdt|�� ���)Nz�target_nodes type can be one of the following: node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. The passed type is )rl r� r� rm rP � TypeError�type)r� r� rj r5 r5 r7 �_parse_target_nodes s
�
����z RedisCluster._parse_target_nodesc
O sN d}d}d}|� dd�}|dur | �|�s | �|�}d}|r dn| j}d| }t|�D ]y} zMi }
|sX| j|i |�d|i��}|sHtd|� d ���t|�dkrX|d | �� krXd}|D ]} | j | g|�R i |��|
| j
< qZ| j |d |
fi |��W S t y� } z |dkr�t
| �| jjv r�|r�| �� |d8 }W Y d} ~ q+| �d} ~ ww dS )
a"
Wrapper for ERRORS_ALLOW_RETRY error handling.
It will try the number of times specified by the config option
"self.cluster_error_retry_attempts" which defaults to 3 unless manually
configured.
If it reaches the number of times, the command will raise the exception
Key argument :target_nodes: can be passed with the following types:
nodes_flag: PRIMARIES, REPLICAS, ALL_NODES, RANDOM
ClusterNode
list<ClusterNode>
dict<Any, ClusterNode>
FNr� Tr r[ r� �!No targets were found to execute � command on)r� r� r� r rn rq r" ro r� �_execute_commandrV �_process_result� Exceptionr� r r r� )
r� r> r� �target_nodes_specified�is_default_noder� �passed_targets�retry_attempts�execute_attempts�_rC ru �er5 r5 r7 �execute_command sR
� ���
� ���z RedisCluster.execute_commandc
O s� |d }d}d}d}d}d} t | j�}
|
dk�rn|
d8 }
�z?zo|r(| j|d�}n| r=| j|� } | j�| | jo9|tv �}d} | �|�}t |g|�R i |��}|r`|�
d� |j |dfi |�� d}|j
|� |j ||fi |��} || j v r| j | | fi |��} | W W |dur�|j
�|� S S ty� � ttfy� }
z|dur�|�� | jj�|jd� d|_| j�� |
�d}
~
w ty� }
z"| jd7 _| �� r�| j�� d| _n| j�|
� d} W Y d}
~
njd}
~
w t y� |
| jd k r�t � d � Y nR t �y }
zt |
j!|
j"d
�}d}W Y d}
~
n:d}
~
w t#�y2 }
z t � d � | j�� |
�d}
~
w t$�y: � t%�yO }
z |�rI|�� |
�d}
~
ww W |du�r\|j
�|� n
|du�ri|j
�|� w w |
dkst&d ��)
z9
Send a command to a node in the cluster
r NFr[ rU �ASKINGTrk g�������?r6 g �?zTTL exhausted.)'r| r� rB rn r� ro r r rA r< r> �parse_responser" r; �releaser r r& r8 r r� rV r7 �
initializer! r rr �update_moved_exceptionr' �timere r r8 r1 r2 r r$ r� r )r� r� r> r� rB r= r: �
redirect_addr�asking�moved�ttlrf rG r� r5 r5 r7 r� ] s�
�
��8� �
�
�
��
����
�
���Vz RedisCluster._execute_commandc C sh z)| j � | jr| j�� W d � W d S W d � W d S 1 s"w Y W d S ty3 Y d S w r@ )r& r� r. �AttributeErrorr* r5 r5 r7 r. � s �&� �zRedisCluster.closec K sB || j v r| j | ||fi |��S t|�dkr t|�� �d S |S )a
Process the result of the executed command.
The function would return a dict or a single value.
:type command: str
:type res: dict
`res` should be in the following format:
Dict<node_name, command_result>
r[ r )r# ro r� rP )r� rB rC r� r5 r5 r7 r� � s
z RedisCluster._process_resultc C s t | ||� dS )a
This function can be used to add externally defined redis modules,
and their namespaces to the redis client.
``funcname`` - A string containing the name of the function to create
``func`` - The function, being added to this class.
N)�setattr)r� �funcname�funcr5 r5 r7 �load_external_module� s z!RedisCluster.load_external_module) Nr
Nr^ NFr FTNN�NNN)F)r� r* r3 Nr@ �NN)Nrc TNNT)0r� r� r� � classmethodr r rd r| r �boolr r r( r+ r3 r5 r; r rA rB r� r� rJ r6 rR r� rW rY rZ r[ r] rb r^ rk rq rr rM rv rn r� r� r� r� r� r� r. r� r� r5 r5 r5 r7 r � s�
-���
����� �
� �
�
�J'
9Fd r c @ s. e Zd Zd
dd�Zdd� Zdd� Zdd � ZdS ) r� Nc C s: |dkr t �|�}|| _|| _t||�| _|| _|| _d S )N� localhost)�socket�
gethostbynamer1 r2 r8 rV � server_typer7 )r� r1 r2 r� r7 r5 r5 r7 r( � s
zClusterNode.__init__c C s. d| j � d| j� d| j� d| j� d| j� d� S )Nz[host=z,port=z,name=z
,server_type=z,redis_connection=�])r1 r2 rV r� r7 r* r5 r5 r7 �__repr__� s
�����zClusterNode.__repr__c C s t |t�o
|j| jkS r@ )rl r� rV )r� �objr5 r5 r7 �__eq__� rK zClusterNode.__eq__c C s | j d ur | j �� d S d S r@ )r7 r. r* r5 r5 r7 r5 s
�zClusterNode.__del__r� )r� r� r� r( r� r� r5 r5 r5 r5 r7 r� � s
r� c @ sD e Zd ZdZddeddfdd�Zded edefd
d �Zdd d
�ZdS )� LoadBalancerz$
Round-Robin Load Balancing
r � start_indexr3 Nc C s i | _ || _d S r@ )�primary_to_idxr� )r� r� r5 r5 r7 r(
s
zLoadBalancer.__init__ra � list_sizec C s& | j �|| j�}|d | | j |< |S )Nr[ )r� �
setdefaultr� )r� ra r� � server_indexr5 r5 r7 �get_server_index s z LoadBalancer.get_server_indexc C s | j �� d S r@ )r� �clearr* r5 r5 r7 �reset r9 zLoadBalancer.reset)r )r3 N) r� r� r� �__doc__r| r( rd r� r� r5 r5 r5 r7 r� s
r� c @ s� e Zd Zddddedfdeeeegeeef f fdd�Z d&dd�Z
d d
� Z d d � Z d'd
d�Z
dd� Zdd� Zdd� Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Zd!ed"ed#eeef fd$d%�ZdS )(r! FNTr c K sv i | _ i | _i | _d | _| �|� || _|| _|| _|| _|| _ d | _
|| _ t � | _
|d u r2t�� }|| _| �� d S r@ )rI rN r r� �populate_startup_nodesr �_require_full_coverage�_dynamic_startup_nodesr} r �_moved_exceptionr� r� �read_load_balancerr% r) r& r� ) r� r r r r^ r r} r r� r5 r5 r7 r( s"
zNodesManager.__init__c C sB |r|r|dkr
t �|�}| j�t||d��S |r | j�|�S dS )z�
Get the requested node from the cluster's nodes.
nodes.
:return: ClusterNode if the node exists, else None
r� r6 N)r� r� rI rc r8 rC r5 r5 r7 rB 6 s
zNodesManager.get_nodec C s
|| _ d S r@ )r� )r� � exceptionr5 r5 r7 r� F s
z#NodesManager.update_moved_exceptionc C s� | j }| j|j|jd�}|dur|jturt|_nt|j|jt�}|| j|j< || j |j
v r]| j |j
d }t |_| j |j
� |� | j |j
�
|� || j |j
d< | j|kr\|| _n|g| j |j
< d| _ dS )z@
Update the slot's node with the redirected one
r6 Nr )r� rB r1 r2 r� rG r� rI rV rN �slot_idrH rp �remover� )r� r� �redirected_node� old_primaryr5 r5 r7 �_update_moved_slotsI s&
�
�
z NodesManager._update_moved_slotsc C s� | j r | j� | j r| �� W d � n1 sw Y | j�|�du s.t| j| �dkr:td|� d| j� d���|du rS| j| d j}| j �
|t| j| ��}n!|du sd|t ksdt| j| �dkrgd}n
t �
dt| j| �d �}| j| | S )z9
Gets a node that servers this hash slot
Nr rL z5" not covered by the cluster. "require_full_coverage=�"Tr[ )r� r& r� rN rc ro r% r� rV r� r� rG r� �randint)r� rf r r� � primary_namerQ r5 r5 r7 ro p s. � �"���z NodesManager.get_node_from_slotc s � fdd�| j �� D �S )z�
Get all nodes with the specified server type
:param server_type: 'primary' or 'replica'
:return: list of ClusterNode
c s g | ] }|j � kr|�qS r5 �r� r� r� r5 r7 rN � s
�z9NodesManager.get_nodes_by_server_type.<locals>.<listcomp>)rI rP )r� r� r5 r� r7 rF � s
�z%NodesManager.get_nodes_by_server_typec C s |D ]}|| j |j< qdS )zK
Populate all startup nodes and filters out any duplicates
N)r rV )r� rj �nr5 r5 r7 r� � s �z#NodesManager.populate_startup_nodesc C s" t dt�D ] }||vr dS qdS )Nr FT)rn r )r� rN rt r5 r5 r7 �check_slots_coverage� s
�z!NodesManager.check_slots_coveragec C s6 |D ]}|j du r| jd|j|jd�| j��|_ qdS )zV
This function will create a redis connection to all nodes in :nodes:
Nr6 r5 )r7 �create_redis_noder1 r2 r� )r� rj ru r5 r5 r7 r@ � s
����z%NodesManager.create_redis_connectionsc K sT | j r |�d|i� |�d|i� t| jdi |��d�}|S td||d�|��}|S )Nr1 r2 )r; r6 r5 )r r r r} )r� r1 r2 r� rI r5 r5 r7 r� � s �z NodesManager.create_redis_nodec C sZ t ||�}|�|�}|d u r+| j�|�}|d u s |jd u r#t|||�}|j|kr+||_|S r@ )r8 rc rI r7 r� r� )r� r1 r2 �role�tmp_nodes_cacherF r� r5 r5 r7 �_get_or_create_cluster_node� s
z(NodesManager._get_or_create_cluster_nodec s | � � i }i }g }d}d}| j}d}| j�� D �]1}z4|jr#|j} n| j|j|jfi |��} | | j|j _t | �
� � d��du rEt d��t
| �d��}
d}W n tyc } z| }W Y d} ~ qd} ~ ww t|
�dkr�t|
d d d �dkr�t| j�dkr�|j|
d d d< |
D ]�� � d } t
| d �}
|
d
kr�|j}
t| d �}| �|
|�\}
}| �|
|t|�}|||j< tt� d �t� d �d �D ]y}||v�rg ||< || �|� � fd d �td
t� ��D �}|D ](}t
|d �}
|d }| �|
|�\}
}| �|
|t|�}|| �|� |||j< q�q�|| d }|j|jk�r>|�|j� d|j� d|� �� t|�dk�r>t dd�|�� ���q�q�| �|�}|�rJ nq|�sXt dt|�� ��|�| �t |�� �� |�su| j �rut dt|�� dt � d���|| _ || _ | �!t�d | _"| j#�r�|| _d| _$dS )z�
Initializes the nodes cache, slots cache and redis connections.
:startup_nodes:
Responsible for discovering other nodes in the cluster
FN�cluster_enabledz(Cluster mode is not enabled on this noder� Tr[ r rk rY c s g | ]}� | �qS r5 r5 )rK �j�rf r5 r7 rN r` z+NodesManager.initialize.<locals>.<listcomp>r^ z vs z
on slot: r z6startup_nodes could not agree on a valid slots cache: z, zORedis Cluster cannot be connected. Please provide at least one reachable node: z9All slots are not covered after query all startup_nodes. z of z covered...)%r� r� r rP r7 r� r1 r2 rV r� �inforc r" r0 r� r� ro r| �remap_host_portr� rG rn rp rH �joinr� rd r@ r� r� r rI rN rF r� r� r� )r� r� � tmp_slots�
disagreements�startup_nodes_reachable�
fully_coveredr� r� � startup_noderI �
cluster_slotsr� � primary_noder1 r2 r� rt �
replica_nodes� replica_node�target_replica_node�tmp_slotr5 r� r7 r� � s� ��� �� �
"
���� �����
"�������
zNodesManager.initializec C s* d | _ | j�� D ]
}|jr|j�� qd S r@ )r� rI rP r7 r. r: r5 r5 r7 r. Z s
��zNodesManager.closec C s& z| j �� W d S ty Y d S w r@ )r� r� r� r* r5 r5 r7 r� ` s
�zNodesManager.resetr1 r2 r3 c C s | j r
| � ||f�S ||fS )z�
Remap the host and port returned from the cluster to a different
internal value. Useful if the client is not connecting directly
to the cluster.
)r )r� r1 r2 r5 r5 r7 r� g s z NodesManager.remap_host_portr� )FN)r� r� r� r r r rd r| r r( rB r� r� ro rF r� r� r@ r� r� r� r. r� r� r5 r5 r5 r7 r! s2 �
�
'#
"r! c s� e Zd ZdZ d � fdd� Zd dd�Zdd� Zd d
� Zd d � Zd
d� Z dd� Z
dd� Z d!dd�Z dd� Z
dd� Zdd � Zd d � Z� ZS )"r\ z�
Wrapper for PubSub class.
IMPORTANT: before using ClusterPubSub, read about the known limitations
with pubsub in Cluster mode and learn how to workaround them:
https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html
Nc sh d| _ | �||||� | j du rdn|�| j �j}|| _i | _| �� | _t� jd||j |d�|�� dS )a*
When a pubsub instance is created without specifying a node, a single
node will be transparently chosen for the pubsub connection on the
first command execution. The node will be determined by:
1. Hashing the channel name in the request to find its keyslot
2. Selecting a node that handles the keyslot: If read_from_replicas is
set to true, a replica can be selected.
:type redis_cluster: RedisCluster
:type node: ClusterNode
:type host: str
:type port: int
N)r; r �push_handler_funcr5 )
ru �set_pubsub_noderA r; �cluster�node_pubsub_mapping�_pubsubs_generator�superr( r )r� �
redis_clusterru r1 r2 r� r� r; �r r5 r7 r( { s
� �
�
�zClusterPubSub.__init__c C s| |dur| � |||j|j� |}n(|dur+|dur+|j||d�}| � ||||� |}nt||g�du r7td��d}|| _dS )aN
The pubsub node will be set according to the passed node, host and port
When none of the node, host, or port are specified - the node is set
to None and will be determined by the keyslot of the channel in the
first command to be executed.
RedisClusterException will be thrown if the passed node does not exist
in the cluster.
If host is passed without port, or vice versa, a DataError will be
thrown.
:type cluster: RedisCluster
:type node: ClusterNode
:type host: str
:type port: int
Nr6 Tz6Passing a host requires passing a port, and vice versa)�_raise_on_invalid_noder1 r2 rB �anyr ru )r� r� ru r1 r2 � pubsub_noder5 r5 r7 r� � s
z ClusterPubSub.set_pubsub_nodec C rX )zJ
Get the node that is being used as the pubsub connection
)ru r* r5 r5 r7 �get_pubsub_node� r� z ClusterPubSub.get_pubsub_nodec C s4 |du s
|j |jd�du rtd|� d|� d���dS )zl
Raise a RedisClusterException if the node is None or doesn't exist in
the cluster.
NrU zNode r4 z doesn't exist in the cluster)rB rV r" )r� r� ru r1 r2 r5 r5 r7 r� � s
��z$ClusterPubSub._raise_on_invalid_nodec G s� | j du rV| jdu r7t|�dkr%|d }| j�|�}| jj�|| jj�}n| j�� }|| _ | j�
|�}|j| _| j� d| j �| _ | j �
| j� | jdurVtsV| j j�| j� | j }| j||jg|�R � dS )z�
Execute a subscribe/unsubscribe command.
Taken code from redis-py and tweak to make it work within a cluster.
Nr[ r] )r: r; ro r� rM r� ro r rJ ru rA r< ra �_register_connect_callbackr r� r+ �_parser�set_push_handler�_executer> )r� r> rL rf ru r7 r: r5 r5 r7 r� � s(
�
�z ClusterPubSub.execute_commandc C sB z| j |j W S ty |jj| jd�}|| j |j< | Y S w )N)r� )r� rV rQ r7 r] r� )r� ru r] r5 r5 r7 �_get_node_pubsub� s � �z ClusterPubSub._get_node_pubsubc C s: t t| j��D ]}t| j�}|�� }|d ur| S qd S r@ )rn ro r� �nextr� � get_message)r� r� r] �messager5 r5 r7 �_sharded_message_generator s
�z(ClusterPubSub._sharded_message_generatorc c s � | j �� D ]}|V qqr@ )r� rP �r� r] r5 r5 r7 r�
s
��z ClusterPubSub._pubsubs_generatorF� c C s� |r| j |j j||d�}n| �� }|d u rd S t|d �dkrP|d | jv rP| j�|d � | j�|d d � | j �
|d �}| j |j j du rP| j �|j� | j s^| j
s^| js^| j�� | jsc|red S |S )N)�ignore_subscribe_messagesrd r� � sunsubscriberL F)r� rV r� r� r0 �"pending_unsubscribe_shard_channelsr� �shard_channelsr� r� rR �
subscribed�channels�patterns�subscribed_eventr� r� )r� r� rd r� r� ru r5 r5 r7 �get_sharded_message s& �
z!ClusterPubSub.get_sharded_messagec O s� |r
t |d |dd � �}t�|�}|�|� |�� D ]A\}}| j�|�}| �|�}|r7|jdi ||i�� n|�|� | j �|j � | j
� | � |d i�� |j
r\| j
s\| j�� d| _qd S )Nr r[ r5 )r rm �fromkeysr rA r� rR r� �
ssubscriber� r� �difference_update�_normalize_keysr� r� �set� health_check_response_counter)r� r> r� �
s_channels� s_channel�handlerru r] r5 r5 r7 r ) s&
�
��zClusterPubSub.ssubscribec G s^ |rt |d |dd � �}n| j}|D ]}| j�|�}| �|�}|�|� | j�|j� qd S rZ )r r� r� rR r� r� r� r )r� r> r ru �pr5 r5 r7 r� = s
��zClusterPubSub.sunsubscribec C s | j dur | j jS dS )zH
Get the Redis connection of the pubsub connected node.
N)ru r7 r* r5 r5 r7 rA K s
�z"ClusterPubSub.get_redis_connectionc C s. | j r| j �� | j�� D ]}|j �� q
dS )z3
Disconnect the pubsub connection.
N)r: r8 r� rP r� r5 r5 r7 r8 R s
�zClusterPubSub.disconnect)NNNNr� )Fr� N)r� r� r� r� r( r� r� r� r� r� r� r� r r r� rA r8 �
__classcell__r5 r5 r� r7 r\ r s( �
'!
%
�r\ c sZ e Zd ZdZeeeeefZ dHdddd d
e
e e e
f d e
e e e
f d e
ed
dededefdd�Zdd� Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Zd!d"� Zd#d$� Zd%d&� ZdId(d)�Z d*d+� Z 'dJd,d-�Z 'dJ� fd.d/� Z d0d1� Z d2d3� Z!d4d5� Z"d6d7� Z#d8d9� Z$d:d;� Z%d<d=� Z&d>d?� Z'd@dA� Z(dBdC� Z)dDdE� Z*dFdG� Z+� Z,S )Kr_ z8
Support for Redis pipeline
in cluster mode
NFr^ r r� r! r$ r
r# r" r r� r r r c
K s� g | _ || _|| _d| _|p| jj�� | _|r|ng | _|| _ | jj
�� | _ || _ || _
d| _|| _t|
�dd�|
�dd�|
�dd��| _| du rMt�� } | | _dS ) rl Fr r� rx r� r r� N)�
command_stackr� r$ �refresh_table_asapr r r r# r r r r r" r r r r rc r r% r) r& ) r� r� r$ r# r" r r r r r^ r� r5 r5 r7 r( j s* �
�
zClusterPipeline.__init__c C s t | �j� S �rl )r� r� r* r5 r5 r7 r� � � zClusterPipeline.__repr__c C r) r r5 r* r5 r5 r7 r+ � � zClusterPipeline.__enter__c C s | � � dS �rl N)r� r/ r5 r5 r7 r3 � r zClusterPipeline.__exit__c C s$ z| � � W d S ty Y d S w r@ )r� r� r* r5 r5 r7 r5 � s
�zClusterPipeline.__del__c C s
t | j�S r )ro r r* r5 r5 r7 �__len__� s
zClusterPipeline.__len__c C s dS )z?Pipeline instances should always evaluate to True on Python 3+Tr5 r* r5 r5 r7 �__bool__� r zClusterPipeline.__bool__c O s | j |i |��S )z?
Wrapper function for pipeline_execute_command
)�pipeline_execute_command�r� r> r� r5 r5 r7 r� � s z ClusterPipeline.execute_commandc O s | j �t||t| j ��� | S )zN
Appends the executed command to the pipeline's command stack
)r rp �PipelineCommandro �r� r> r? r5 r5 r7 r � s �z(ClusterPipeline.pipeline_execute_commandc C s8 |D ]}|j }t|t�r| �||jd |j� |�qdS )z8
Raise the first exception on the stack
r[ N)�resultrl r� �annotate_exception�positionr> )r� �stack�crI r5 r5 r7 �raise_first_error� s
��z!ClusterPipeline.raise_first_errorc C sF d� tt|��}d|� d|� d|jd � �}|f|jdd� |_dS )zS
Provides extra context to the exception prior to it being handled
rl z
Command # z (z ) of pipeline caused error: r r[ N)r� �mapr/ r> )r� r� �numberrB �cmd�msgr5 r5 r7 r � s ��z"ClusterPipeline.annotate_exceptionTc C s( | j }z | �||�W | �� S | �� w )zB
Execute all the commands in the current pipeline
)r �send_cluster_commandsr� )r� �raise_on_errorr r5 r5 r7 �execute� s zClusterPipeline.executec C s g | _ t� | _d| _d| _dS )z/
Reset back to empty pipeline.
FN)r r �scripts�watching�explicit_transactionr* r5 r5 r7 r� � s
zClusterPipeline.resetc
C sd |sg S | j } z | j|||d�W S ttfy0 } z|dkr$|d8 }n|�W Y d}~nd}~ww q)a
Wrapper for CLUSTERDOWN error handling.
If the cluster reports it is down it is assumed that:
- connection_pool was disconnected
- connection_pool was reseted
- refereh_table_asap set to True
It will try the number of times specified by
the config option "self.cluster_error_retry_attempts"
which defaults to 3 unless manually configured.
If it reaches the number of times, the command will
raises ClusterDownException.
T)r# �allow_redirectionsr r[ N)r �_send_cluster_commandsr r )r� r r# r( r� r� r5 r5 r7 r" � s& � ����z%ClusterPipeline.send_cluster_commandsc sx t |dd� d�}d}i }|D ]} |j�dd�}|r%| �|�s%| �|�} n| j|jd|i�} | s9td |j� d
���t| �d krGtd |j� ���| d
}
|
| � � krSd}|
j
} | |vr�| � |
�} zt | |j�}
W n t
yz | j�� |ry| �� � w t| j| j|
�|| < || �|� q|�� }|D ]}|�� q�|D ]}|�� q�|�� D ] }|j�|j� q�t dd� |D �dd� d�}|�r|�r| jd 7 _| �� r�| j�� |r�| �� |D ]&}zt � j |ji |j��|_ W q� t �y } z ||_ W Y d}~q�d}~ww g }t |dd� d�D ]#}|jd
| j v �r*| j |jd
|j fi |j��|_ |�|j � �q|�r:| �!|� |S )z�
Send a bunch of cluster commands to the redis cluster.
`allow_redirections` If the pipeline should follow
`ASK` & `MOVED` responses automatically. If set
to false it will raise RedisClusterException.
c S rX r@ �r �rr r5 r5 r7 r� * � z8ClusterPipeline._send_cluster_commands.<locals>.<lambda>)rO FTr� N� node_flagr� r� r[ z Too many targets for command r c s s" � | ] }t |jtj�r|V qd S r@ )rl r r_ r �rK r r5 r5 r7 � <genexpr>� s � �
�z9ClusterPipeline._send_cluster_commands.<locals>.<genexpr>c S rX r@ r* r+ r5 r5 r7 r� � r, c S rX r@ r* r+ r5 r5 r7 r� � r, )"�sortedr? r� r� r� rq r> r" ro r� rV rA r< r r� r� r� � NodeCommandsr� r; rp rP �write�readr� r: r rr r� r� r r# r" r )r� r r# r( �attemptr� rj r r� r� ru rF r= r: �
node_commandsr� r� rG r� r5 r7 r) s� �� �
�
��
��
����
z&ClusterPipeline._send_cluster_commandsc C s |st d��dS )rl z4ASK & MOVED redirection not allowed in this pipelineN�r" )r� r( r5 r5 r7 �_fail_on_redirect� s
��z!ClusterPipeline._fail_on_redirectc G s | j dg|�R � S )N�EXISTS)r� )r� r� r5 r5 r7 �exists� s zClusterPipeline.existsc C � t d��)rl z method eval() is not implementedr6 r* r5 r5 r7 �eval� � zClusterPipeline.evalc C r: )rl z!method multi() is not implementedr6 r* r5 r5 r7 �multi� r< zClusterPipeline.multic O r: )rl z5method immediate_execute_command() is not implementedr6 r r5 r5 r7 �immediate_execute_command� � �z)ClusterPipeline.immediate_execute_commandc O r: )rl z0method _execute_transaction() is not implementedr6 r r5 r5 r7 �_execute_transaction� r< z$ClusterPipeline._execute_transactionc C r: )rl z(method load_scripts() is not implementedr6 r* r5 r5 r7 � load_scripts� r< z ClusterPipeline.load_scriptsc G r: )rl z!method watch() is not implementedr6 �r� �namesr5 r5 r7 �watch� r< zClusterPipeline.watchc C r: )rl z#method unwatch() is not implementedr6 r* r5 r5 r7 �unwatch� r< zClusterPipeline.unwatchc O r: )rl z4method script_load_for_pipeline() is not implementedr6 r r5 r5 r7 �script_load_for_pipeline� r? z(ClusterPipeline.script_load_for_pipelinec G �$ t |�dkr
td��| �d|d �S )z7
"Delete a key specified by ``names``"
r[ z=deleting multiple keys is not implemented in pipeline command�DELr �ro r" r� rB r5 r5 r7 �delete� �
�zClusterPipeline.deletec G rG )z7
"Unlink a key specified by ``names``"
r[ z>unlinking multiple keys is not implemented in pipeline command�UNLINKr rI rB r5 r5 r7 �unlink� rK zClusterPipeline.unlink)NNNFr^ r N)T)TT)-r� r� r� r� r r& r! r r' r r r rd r r r� r| r( r� r+ r3 r5 r r r� r r r r$ r� r" r) r7 r9 r; r= r> r@ rA rD rE rF rJ rM r r5 r5 r� r7 r_ \ sz � �����
���
�%
!
�&� ) r_ rV .c s � fdd�}|S )zi
Prints error because some pipelined commands should
be blocked when running in cluster-mode
c s t d� � d���)Nz"ERROR: Calling pipelined function z1 is blocked when running redis in cluster mode...r6 )r> r� �rV r5 r7 �inner s
�z%block_pipeline_command.<locals>.innerr5 )rV rO r5 rN r7 �block_pipeline_command s rP )E� BGREWRITEAOFr� �BITOP�
BRPOPLPUSHr� r� r� r� �CLIENTr� r� r� r� �CONFIGr� �ECHOrx r� r� r� r� r� �MGETzMGET NONATOMIC�MOVE�MSETzMSET NONATOMIC�MSETNX�PFCOUNT�PFMERGEr� �PUBLISHr� r� r� �RENAME�RENAMENX� RPOPLPUSHr� r� r� r� z SCRIPT KILLr� �SCRIPT�SDIFF�
SDIFFSTOREz SENTINEL GET MASTER ADDR BY NAMEzSENTINEL MASTERzSENTINEL MASTERSzSENTINEL MONITORzSENTINEL REMOVEzSENTINEL SENTINELSz SENTINEL SETzSENTINEL SLAVES�SENTINELr� �SINTER� SINTERSTORE�SLAVEOFr� r� r� �SLOWLOG�SMOVE�SORT�SUNION� SUNIONSTOREr� rl r� c @ s e Zd ZdZddd�ZdS )r rl Nc C s4 || _ |d u r i }|| _|| _d | _d | _d| _d S )NF)r> r? r r ru r� )r� r> r? r r5 r5 r7 r( i s
zPipelineCommand.__init__r� )r� r� r� r� r( r5 r5 r5 r7 r f s r c @ s0 e Zd ZdZdd� Zdd� Zdd� Zdd � Zd
S ) r1 rl c C s || _ || _|| _g | _dS r )r� r; r: �commands)r� r� r; r: r5 r5 r7 r( w s
zNodeCommands.__init__c C s | j �|� dS r )rm rp )r� r r5 r5 r7 rp ~ s zNodeCommands.appendc
C sx | j }| j}|D ]}d|_qz|�|�dd� |D ��� W dS ttfy; } z|D ]}||_q*W Y d}~dS d}~ww )z=
Code borrowed from Redis so it can be fixed
Nc S s g | ]}|j �qS r5 r\ r. r5 r5 r7 rN � s z&NodeCommands.write.<locals>.<listcomp>)r: rm r �send_packed_command�
pack_commandsr r& )r� r: rm r r� r5 r5 r7 r2 � s
����zNodeCommands.writec C s� | j }| jD ]G}|jdu rMz| j||jd fi |j��|_W q ttfy= } z| jD ]}||_q+W Y d}~ dS d}~w tyL t �
� d |_Y qw qdS )rl Nr r[ ) r: rm r r� r> r? r r& r# �sys�exc_info)r� r: r r� r5 r5 r7 r3 � s
"
� ���zNodeCommands.readN)r� r� r� r� r( rp r2 r3 r5 r5 r5 r7 r1 t s r1 )`r� r� rp r% r� � collectionsr �typingr r r r r r r �redis._parsersr
r �redis._parsers.helpersr �
redis.backoffr
� redis.clientr r r �redis.commandsr r �redis.commands.helpersr �redis.connectionr r r � redis.crcr r �redis.exceptionsr r r r r r r r r! r" r# r$ r% r& r' �
redis.lockr) � redis.retryr* � redis.utilsr+ r, r- r. r/ r0 rd r| r8 r< rJ rV ri rw rz rG rH r r� r� r� r� r� r r� r� r! r\ r_ rP �PIPELINE_BLOCKED_COMMANDSrB �replace�lowerr� r r1 r5 r5 r5 r7 �<module> s� $ D
��
�&
u 2 \ k 0G