U d N@sddlZddlZddlZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z ddlmZmZmZmZddlmZmZmZddlmZmZmZmZddlmZmZddlm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/dd l0m1Z1dd l2m3Z3m4Z4m5Z5m6Z6m7Z7e8e9Z:e;eddZ?ddZ@e e e e eKsz'parse_pubsub_numsub..)rvaluesKeyErrorr:)r;r<r8Znumsub_dZ numsub_tupsrDZ numsubbedZ ret_numsubr/r/r1parse_pubsub_numsubBs  rH)respr8r-c s||ddttttfdfdd i}|D]F}|dd\}}}|dd}|fdd|Dd |||f<q0|S) N current_host)r7r-cst|dp|dfSNr)r*r7)rJr/r1 fix_serverTsz'parse_cluster_slots..fix_servercsg|] }|qSr/r/)rCreplica)rOr/r1rE]sz'parse_cluster_slots..)primaryreplicas)getrrstr)rIr8slotsslotstartendrRrSr/)rJrOr1parse_cluster_slotsOs  rZc Ksg}|D]}ggd}tdt|ddD]*}|d|d||d|dfq*|d}|D]@}i}tdt|dD]}||d|||<qz|d|qb||q|S)z( Parse CLUSTER SHARDS response. )rVnodesrrMrVrPr[)rangelenappend) rIr8ZshardsxZshardir[nodeZ dict_noder/r/r1parse_cluster_shardscs ( rcrRrQzslot-id)charsetZconnection_classr4Z client_namedbdecode_responsesencodingencoding_errorserrorsr+Zmax_connections nodes_flagredis_connect_funcpasswordr,retryZretry_on_timeoutZsocket_connect_timeoutZsocket_keepaliveZsocket_keepalive_optionssocket_timeoutsslZ ssl_ca_certsZ ssl_ca_dataZ ssl_certfileZ ssl_cert_reqsZ ssl_keyfileZ ssl_passwordZunix_socket_pathusernamer0cKsdd|D}|S)z9 Remove unsupported or disabled keys from kwargs cSs&i|]\}}|tkr|tkr||qSr/)REDIS_ALLOWED_KEYSKWARGS_DISABLED_KEYS)rCkvr/r/r1 s z"cleanup_kwargs..)r:)kwargsconnection_kwargsr/r/r1cleanup_kwargssrxc @s&eZdZeejeeee e e dZdS) ClusterParser)ZASKZTRYAGAINMOVED CLUSTERDOWNZ CROSSSLOT MASTERDOWNN) __name__ __module__ __qualname__r&rZEXCEPTION_CLASSESrr#rrrrr/r/r/r1rysrycI@seZdZdZdZdZdZdZdZeeeeehZ e e ddd d d d d ddddddddddddddddddd d!d"d#d$d%d&d'd(d)d*d+d,d-d.d/d0d1d2d3d4d5d6d7d8d9d:d;dd?d@dAdBdCdDdEdFdGdHdId7d8ddJgGee dKdLdMdNdOdPdQdRdSdTdUg ee dVgee dWdXdYdZd[ge Z d\d]d^d_d`dadbdcdddedfdgdhdidjdkdldmdndodpdqdrdsdtdudvdwdxdydzgfZeed{Ze e dgee dgd|d}e d"dgee dddddd$d'd)d*d/d0g d~d}e d#d(gdd}e d1gdd}e dRgee dUgdd}e dSgdd}e dTgdd} ZeeefZdS)AbstractRedisClusterZ primariesrSallrandomz default-nodezACL CATz ACL DELUSERz ACL DRYRUNz ACL GENPASSz ACL GETUSERzACL HELPzACL LISTzACL LOGzACL LOADzACL SAVEz ACL SETUSERz ACL USERSz ACL WHOAMIZAUTH CLIENT LISTCLIENT SETNAMECLIENT GETNAME CONFIG SETCONFIG REWRITECONFIG RESETSTATTIMEzPUBSUB CHANNELSz PUBSUB NUMPATz PUBSUB NUMSUBPINGINFOSHUTDOWNKEYSDBSIZEBGSAVE SLOWLOG GET SLOWLOG LEN SLOWLOG RESETZWAITSAVEz MEMORY PURGEzMEMORY MALLOC-STATSz MEMORY STATSLASTSAVEzCLIENT TRACKINGINFOz CLIENT PAUSEzCLIENT UNPAUSEzCLIENT UNBLOCKz CLIENT IDz CLIENT REPLYzCLIENT GETREDIRz CLIENT INFO CLIENT KILLREADONLY READWRITEz CLUSTER INFOz CLUSTER MEETz CLUSTER NODESzCLUSTER REPLICASz CLUSTER RESETzCLUSTER SET-CONFIG-EPOCH CLUSTER SLOTSCLUSTER SHARDSzCLUSTER COUNT-FAILURE-REPORTSzCLUSTER KEYSLOTZCOMMANDz COMMAND COUNTz COMMAND LISTzCOMMAND GETKEYS CONFIG GETDEBUG RANDOMKEYz GRAPH.CONFIGFLUSHALLFLUSHDBzFUNCTION DELETEzFUNCTION FLUSHz FUNCTION LISTz FUNCTION LOADzFUNCTION RESTORESCAN SCRIPT EXISTS SCRIPT FLUSH SCRIPT LOADz FUNCTION DUMPzCLUSTER COUNTKEYSINSLOTzCLUSTER DELSLOTSzCLUSTER DELSLOTSRANGEzCLUSTER GETKEYSINSLOTzCLUSTER SETSLOTz 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)rrcCstt|Sr9)sumlistrFr;r<r/r/r1ZzAbstractRedisCluster.cCst|trt|S|Sr9) isinstancedictrrFrr/r/r1rkrcCst|trt|S|Sr9)rrrrFrr/r/r1rorcCst|dkrdSdSrL)rrFrr/r/r1rrrcCst|Sr9)rrFpoprr/r/r1rvrcCsddt|DS)NcSsg|] }t|qSr/)r)rCrsr/r/r1rEysz1AbstractRedisCluster...)ziprFrr/r/r1ryrcCs t|Sr9)rrFrr/r/r1r{rN)r}r~rRedisClusterRequestTTL PRIMARIESREPLICAS ALL_NODESRANDOM DEFAULT_NODE NODE_FLAGSr&r'SLOT_ID COMMAND_FLAGSSEARCH_COMMANDSrZrc#CLUSTER_COMMANDS_RESPONSE_CALLBACKSrHr(rBRESULT_CALLBACKSrr"rERRORS_ALLOW_RETRYr/r/r/r1rsvIL  l%    'rc @s6eZdZeddZdNd d Zd d Zd dZddZddZ ddZ ddZ dOddZ ddZ ddZddZdd ZdPd!d"Zd#d$Zd%d&ZdQd'd(ZdRd)d*ZdSd+d,ZdTd.d/Zd0d1Zd2d3Zd4d5Zd6d7Zd8d9Zd:d;Zdd?Zd@dAZ dBdCZ!dDdEZ"dFdGZ#dHdIZ$dJdKZ%dLdMZ&dS)U RedisClustercKs|fd|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]:[password]]@/path/to/socket.sock?db=0 Three URL schemes are supported: - `redis://` creates a TCP socket connection. See more at: - `rediss://` creates a SSL wrapped TCP socket connection. See more at: - ``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. urlr/)clsrrvr/r/r1from_urls*zRedisCluster.from_urlNrPF Tc Ks|dkr g}d| krtdd} | dk rd} t| } d| krDtdd| kr`| ddkr`td | | | d }| d |}|t||n6|dk r|dk r|t||nt|dkrtd td || dd|_ | d|j it f| } t | dd| dd| dd|_||_|jj|_|jj|_||_d|_||_d|_tf|| ||d| |_t|jj|_t|jj|_t ||_!t"#|_$dS)a Initialize a new RedisCluster client. :startup_nodes: 'list[ClusterNode]' List of nodes from which initial bootstrapping can be done :host: 'str' Can be used to point to a startup node :port: 'int' Can be used to point to a startup node :require_full_coverage: 'bool' 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. :read_from_replicas: 'bool' 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. :dynamic_startup_nodes: 'bool' 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. :cluster_error_retry_attempts: 'int' Retry command execution attempts when encountering ClusterDownError or ConnectionError :reinitialize_steps: 'int' 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. :**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) Nrez4Argument 'db' is not possible to use in cluster modeFTpathzFRedisCluster does not currently support Unix Domain Socket connectionsrz9A ``db`` querystring option can only be 0 in cluster moder+r,a5RedisCluster 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)])zstartup_nodes : rkrgutf-8rhstrictrf) startup_nodesrrequire_full_coveragedynamic_startup_nodes)%rrupdaterTr_ ClusterNoder^logdebugruser_on_connect_func on_connectrxrencodercluster_error_retry_attempts __class__rcopy command_flagsr node_flagsread_from_replicasreinitialize_counterreinitialize_steps nodes_manager NodesManagerrrcluster_response_callbacksrresult_callbacksr commands_parser threadingr%_lock) selfr+r,rrrrrrrrvrZ url_optionsr/r/r1__init__stC          zRedisCluster.__init__cCs|Sr9r/rr/r/r1 __enter__?szRedisCluster.__enter__cCs |dSr9closerexc_type exc_value tracebackr/r/r1__exit__BszRedisCluster.__exit__cCs |dSr9rrr/r/r1__del__EszRedisCluster.__del__c Cs>|D]0}|jrz|jjWqtk r6YqXqdSr9) get_nodesredis_connectionr4 disconnectOSErrorrrbr/r/r1disconnect_connection_poolsHs  z(RedisCluster.disconnect_connection_poolscCsR|t||jr:|dt|dkr:td|jdk rN||dS)z Initialize the connection, authenticate and select a database and send READONLY if it is set during object initialization. rOKzREADONLY command failedN) set_parserryrr send_commandr*Z read_responserr)rr3r/r/r1rQs   zRedisCluster.on_connectc Cs2|js,|j|js"|j|gW5QRX|jSr9)rrrcreate_redis_connectionsrr/r/r1get_redis_connectionfs z!RedisCluster.get_redis_connectioncCs|j|||Sr9)rget_noderr+r,r>r/r/r1rmszRedisCluster.get_nodecCs |jtSr9)rget_nodes_by_server_typePRIMARYrr/r/r1 get_primariespszRedisCluster.get_primariescCs |jtSr9)rrREPLICArr/r/r1 get_replicassszRedisCluster.get_replicascCstt|jjSr9)rchoicerr nodes_cacherFrr/r/r1get_random_nodevszRedisCluster.get_random_nodecCst|jjSr9)rrrrFrr/r/r1ryszRedisCluster.get_nodescCsn||}|jj|}|dks,t|dkrzVThe requested node does not exist in the cluster, so the default node was not changed.Fz$Changed the default cluster node to T)rnamerinforrrr/r/r1set_default_nodeszRedisCluster.set_default_nodecCs6|dkr|}|jdkr,td|jd|jS)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)rrrrmonitor)r target_noder/r/r1r s   zRedisCluster.monitorcKst|f|||d|S)z~ Allows passing a ClusterNode, or host&port, to get a pubsub instance connected to the specified node )rbr+r,) ClusterPubSub)rrbr+r,rvr/r/r1pubsubszRedisCluster.pubsubc CsF|r td|rtdt|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) rrrrrrrrlock) rClusterPipelinerrrrrrrrr)rZ transaction shard_hintr/r/r1pipelines zRedisCluster.pipeline皙?c Cs"|dkr t}||||||||dS)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)timeoutsleepblockingblocking_timeout thread_localr$)rrrrrr lock_classrr/r/r1r s>zRedisCluster.lockcCs||j|<dS)zSet a custom Response CallbackN)r)rr;callbackr/r/r1set_response_callback#sz"RedisCluster.set_response_callbackcOs\|d}t|dkrR|dd|d|jkrR|dd|d}|dd}|dk rl|}n |j|}|rtd|d|||jjkr| gS||jj kr| S||jj kr| S||jjkr|S||jjkr|jjgS||jjdkr|jjgS|j|}|j||jo8|tk}td|d ||gSdS) Nrr\ rMrjzTarget node/s for z: z Target for z: slot )upperr^rrrTrrrrrrrrrrrrrrrdetermine_slotget_node_from_slotrr )rr7rvr;rjZ command_flagrWrbr/r/r1_determine_nodes's8 ,          zRedisCluster._determine_nodescCs"|jdkrdS|j|jdkSdS)NrF)rrrr/r/r1_should_reinitializedOs z"RedisCluster._should_reinitializedcCs|j|}t|S)z Calculate keyslot for a given key. See Keys distribution model in https://redis.io/topics/cluster-spec )rencoder)rrrsr/r/r1rYs zRedisCluster.keyslotcGs|j}|jj|f|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. )rrrZget_keys)rr7Z redis_connr/r/r1_get_command_keysas zRedisCluster._get_command_keyscs|d}j|tkr |dS|dkrxt|dkrBtd||d}|dd|}t|dkrrtdtS|}n@j|}|dkst|dkr|dkrtdtStd |t|dkr҈ |dSfd d |D}t|dkrt|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. rrM)ZEVALEVALSHAr\zInvalid args in command: rPN)ZFCALLZFCALL_ROzNo way to dispatch this command to Redis Cluster. Missing key. You can execute the command by specifying target nodes. Command: csh|]}|qSr/)r)rCrrr/r1 sz.RedisCluster.determine_slot..z) - all keys must map to the same key slot) rrTrr^rr randrangerr!rr)rr7r;Znum_actual_keysZ eval_keyskeysrVr/rr1rps4       zRedisCluster.determine_slotcCs|jS)z. Get the connections' encoder )rrr/r/r1 get_encoderszRedisCluster.get_encodercCs|jjS)z9 Get the connections' key-word arguments )rrwrr/r/r1get_connection_kwargssz"RedisCluster.get_connection_kwargscCst|to||jkSr9)rrUr)r target_nodesr/r/r1_is_nodes_flagszRedisCluster._is_nodes_flagcCsLt|tr|}n8t|tr"|g}n&t|tr6|}ntdt||S)Nztarget_nodes type can be one of the following: node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),ClusterNode, list, or dict. The passed type is )rrrrrF TypeErrortype)rr(r[r/r/r1_parse_target_nodess     z RedisCluster._parse_target_nodesc Os d}d}|dd}|dk r4||s4||}d}|r dict FNr(TrMrrj!No targets were found to execute command on)rr)r,rr]rr_execute_commandr_process_result BaseExceptionr+rr) rr7rvZtarget_nodes_specifiedr(passed_targetsZretry_attempts exception_r<rber/r/r1execute_commands>    zRedisCluster.execute_commandc Os|d}d}d}d}d}d} t|j} d} | dkr| d8} z8z|rT|j|d}n*| r~|j|} |j| |jov|t k}d} t d|d|j d|j ||}t|f||}|r|d |j|d f|d}|j||j||f|} ||jkr|j|| f|} | WW\Stttfk rX}zt t|W5d}~XYnttfk r}zTt t||dk r|| d7} | d krtd nd|_|jW5d}~XYntk r8}zHt d |j d7_ |!r|jd|_ n |j"|d } W5d}~XYn<t#k rrt d| |jdkrltdYnt$k r}z"t dt%|j&|j'd}d }W5d}~XYnt(k r}z&t dtd |j|W5d}~XYn~t)k r4}z |*}t d||W5d}~XYn@t+k rr}z t d|r^||W5d}~XYnXW5|dk r|j|Xq*t,ddS)z9 Send a command to a node in the cluster rNFrMrzExecuting command z on target node: rZASKINGg?rTr#r\g?rr0rzResponseError: r1zTTL exhausted.)-intrr4releaserrrrrr rr server_typerrr5rparse_responserrrrr3r+rr"rtimerr initializerrrupdate_moved_exceptionr#rr2r+r,rr __str__r1r)rr r7rvr;r6r3Z redirect_addraskingmovedZttlZconnection_error_retry_counterrWr?r5messager/r/r1r/ s                        zRedisCluster._execute_commandc Cs@z&|j|jr|jW5QRXWntk r:YnXdSr9)rrrAttributeErrorrr/r/r1rs zRedisCluster.closecKsB||jkr|j|||f|St|dkr:t|dS|SdS)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 rMrN)rr^rrF)rr;r<rvr/r/r1r0s  zRedisCluster._process_resultcCst|||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)rfuncnamefuncr/r/r1load_external_modulesz!RedisCluster.load_external_module) NrNrPFrFTN)NNN)F)N)NNN)NN)NrTNNT)'r}r~r classmethodrrrrrrrrrrrrrrrrr r rr rrrrr!rr&r'r)r,r6r/rr0rGr/r/r/r1rsh -        J( 9?y rc@s.eZdZd ddZddZddZdd ZdS) rNcCs:|dkrt|}||_||_t|||_||_||_dS)N localhost)socket gethostbynamer+r,r2rr:r)rr+r,r:rr/r/r1rs  zClusterNode.__init__c Cs.d|jd|jd|jd|jd|jd S)Nz[host=z,port=z,name=z ,server_type=z,redis_connection=])r+r,rr:rrr/r/r1__repr__s,zClusterNode.__repr__cCst|to|j|jkSr9)rrr)robjr/r/r1__eq__szClusterNode.__eq__cCs|jdk r|jdSr9)rrrr/r/r1rs zClusterNode.__del__)NN)r}r~rrrMrOrr/r/r/r1rs  rc@sBeZdZdZd eddddZeeeddd Zdd d d ZdS) LoadBalancerz$ Round-Robin Load Balancing rN) start_indexr-cCsi|_||_dSr9)primary_to_idxrQ)rrQr/r/r1rszLoadBalancer.__init__)rR list_sizer-cCs&|j||j}|d||j|<|S)NrM)rR setdefaultrQ)rrRrSZ server_indexr/r/r1get_server_indexszLoadBalancer.get_server_index)r-cCs|jdSr9)rRclearrr/r/r1resetszLoadBalancer.reset)r) r}r~r__doc__r8rrUrUrWr/r/r/r1rPsrPc@seZdZd ddZd!ddZdd Zd d Zd"d d ZddZddZ ddZ ddZ ddZ ddZ ddZddZddZdS)#rFNTcKsji|_i|_i|_d|_||||_||_||_d|_||_ t |_ |dkrXt }||_|dSr9)rrrrpopulate_startup_nodesr_require_full_coverage_dynamic_startup_nodes_moved_exceptionrwrPread_load_balancerrr%rr=)rrrrr rrvr/r/r1rs  zNodesManager.__init__cCsP|r.|r.|dkrt|}|jt||dS|r>|j|StddSdS)z Get the requested node from the cluster's nodes. nodes. :return: ClusterNode if the node exists, else None rIr0zEget_node requires one of the following: 1. node name 2. host and portN)rJrKrrTr2rerrorrr/r/r1rs  zNodesManager.get_nodecCs ||_dSr9)r\)rr3r/r/r1r> sz#NodesManager.update_moved_exceptioncCs|j}|j|j|jd}|dk r2|jtk rNt|_nt|j|jt}||j|j<||j |j kr|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 r0Nr)r\rr+r,r:rrrrrZslot_idrr_remover)rr5Zredirected_nodeZ old_primaryr/r/r1_update_moved_slots s"   z NodesManager._update_moved_slotsc Cs|jr&|j|jr|W5QRX|j|dksHt|j|dkr`td|d|jd|dkr|j|dj}|j |t|j|}nB|dks|t kst|j|dkrd}nt dt|j|d}|j||S)z9 Gets a node that servers this hash slot Nrrz5" not covered by the cluster. "require_full_coverage="TrM)r\rr`rrTr^r!rZrr]rUrrrandint)rrWrr:Z primary_namerr/r/r1r4s." zNodesManager.get_node_from_slotcsfdd|jDS)z Get all nodes with the specified server type :param server_type: 'primary' or 'replica' :return: list of ClusterNode csg|]}|jkr|qSr/r:)rCrbrcr/r1rE]s z9NodesManager.get_nodes_by_server_type..)rrF)rr:r/rcr1rWs z%NodesManager.get_nodes_by_server_typecCs|D]}||j|j<qdS)zK Populate all startup nodes and filters out any duplicates N)rr)rr[nr/r/r1rYcsz#NodesManager.populate_startup_nodescCs"tdtD]}||kr dSq dS)NrFT)r]r)rrrar/r/r1check_slots_coveragejsz!NodesManager.check_slots_coveragecCs6|D],}|jdkr|jf|j|jd|j|_qdS)zV This function will create a redis connection to all nodes in :nodes: Nr0)rcreate_redis_noder+r,rw)rr[rbr/r/r1rrs z%NodesManager.create_redis_connectionscKsL|jr4|d|i|d|ittf|d}ntf||d|}|S)Nr+r,)r4r0)rrr r)rr+r,rvrAr/r/r1rf|s zNodesManager.create_redis_nodecCsJt||}||}|dkrF|j|}|dks:|jdkrFt|||}|Sr9)r2rTrrr)rr+r,roletmp_nodes_cacher>r r/r/r1_get_or_create_cluster_nodes    z(NodesManager._get_or_create_cluster_nodec std|i}i}g}d}d}|j}|jD]4}z~|jrL|j}n.rPz vs z on slot: r7z6startup_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...)-rrrWrwrrFrrdeepcopyrrfr+r,rboolrrTrr*r6rr"r?r3r Exceptionr^r8rirr]r_rjoinrerrrZrrrrrr[r\)rrhZ tmp_slotsZ disagreementsZstartup_nodes_reachableZ fully_coveredrvZ startup_noderAZ copy_kwargsZ cluster_slotsr5msgrBZ primary_noder+r,r raZ replica_nodesZ replica_nodeZtarget_replica_nodeZtmp_slotr/rkr1r=s           "     zNodesManager.initializecCs*d|_|jD]}|jr|jqdSr9)rrrFrrrr/r/r1r3szNodesManager.closecCs(z|jWntk r"YnXdSr9)r]rWr*rr/r/r1rW9szNodesManager.reset)FFNT)NNN)FN)r}r~rrrr>r`rrrYrerrfrir=rrWr/r/r/r1rs&  ' #   rcsLeZdZdZdfdd ZdddZddZd d Zd d Zd dZ Z S)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 sVd|_||||||jdkr$dn ||jj}||_tjf|||jddS)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)r4r)rbset_pubsub_noderr4clustersuperrr)r redis_clusterrbr+r,rvr4rr/r1rJs zClusterPubSub.__init__cCs~|dk r"||||j|j|}nR|dk rV|dk rV|j||d}||||||}nt||gdkrptdnd}||_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 Nr0Tz6Passing a host requires passing a port, and vice versa)_raise_on_invalid_noder+r,ranyrrb)rrrrbr+r,Z pubsub_noder/r/r1rqds zClusterPubSub.set_pubsub_nodecCs|jS)zJ Get the node that is being used as the pubsub connection )rbrr/r/r1get_pubsub_nodeszClusterPubSub.get_pubsub_nodecCs4|dks|j|jddkr0td|d|ddS)zl Raise a RedisClusterException if the node is None or doesn't exist in the cluster. NrzNode r.z doesn't exist in the cluster)rrr)rrtrbr+r,r/r/r1rvsz$ClusterPubSub._raise_on_invalid_nodecOs|jdkr|jdkrnt|dkrJ|d}|j|}|jj||jj}n |j}||_ |j |}|j|_|j d|j |_|j |j|j}|j||jf|dS)z Execute a publish/subscribe command. Taken code from redis-py and tweak to make it work within a cluster. NrMr )r3r4r^rrrrrrrrbrr5rZregister_connect_callbackrZ_executer)rr7rvrDrWrbrr3r/r/r1r6s(      zClusterPubSub.execute_commandcCs|jdk r|jjSdS)zH Get the Redis connection of the pubsub connected node. N)rbrrr/r/r1rs z"ClusterPubSub.get_redis_connection)NNN)NNN) r}r~rrXrrqrxrvr6r __classcell__r/r/rur1r As ! #r cseZdZdZeeeeefZ d=ddZ dd Z d d Z d d Z ddZddZddZddZddZddZddZddZd>dd Zd!d"Zd?d#d$Zd@fd%d& Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0Zd1d2Zd3d4Z d5d6Z!d7d8Z"d9d:Z#d;d<Z$Z%S)Arz8 Support for Redis pipeline in cluster mode NFr7rc Ksg|_||_||_d|_|p&|jj|_|r2|ng|_||_ |jj |_ ||_ ||_ d|_||_t| dd| dd| dd|_| dkrt} | |_dS) rFrrgrrhrrfN) command_stackrrZrefresh_table_asaprrrrrrrrrrrrrrTrrr%r) rrrrrrrrrr rvr/r/r1rs*   zClusterPipeline.__init__cCs t|jSr)r+r}rr/r/r1rMszClusterPipeline.__repr__cCs|Sr{r/rr/r/r1rszClusterPipeline.__enter__cCs |dSrN)rWrr/r/r1rszClusterPipeline.__exit__cCs&z |Wntk r YnXdSr9)rWrnrr/r/r1rs zClusterPipeline.__del__cCs t|jSr{)r^rzrr/r/r1__len__szClusterPipeline.__len__cCsdS)z@Pipeline instances should always evaluate to True on Python 2.7Tr/rr/r/r1 __nonzero__ szClusterPipeline.__nonzero__cCsdS)z?Pipeline instances should always evaluate to True on Python 3+Tr/rr/r/r1__bool__ szClusterPipeline.__bool__cOs |j||S)z? Wrapper function for pipeline_execute_command )pipeline_execute_commandrr7rvr/r/r1r6szClusterPipeline.execute_commandcOs|jt||t|j|S)zN Appends the executed command to the pipeline's command stack )rzr_PipelineCommandr^rr7r8r/r/r1rsz(ClusterPipeline.pipeline_execute_commandcCs8|D].}|j}t|tr|||jd|j|qdS)z8 Raise the first exception on the stack rMN)resultrrnannotate_exceptionpositionr7)rstackcrAr/r/r1raise_first_error s  z!ClusterPipeline.raise_first_errorcCsFdtt|}d|d|d|jd}|f|jdd|_dS)zS Provides extra context to the exception prior to it being handled rz Command # z (z) of pipeline caused error: rrMN)romapr)r7)rr3numberr;cmdrpr/r/r1r*sz"ClusterPipeline.annotate_exceptionTcCs&|j}z|||WS|XdS)zB Execute all the commands in the current pipeline N)rzrWsend_cluster_commands)rraise_on_errorrr/r/r1execute5szClusterPipeline.executecCsg|_t|_d|_d|_dS)z/ Reset back to empty pipeline. FN)rzsetscriptsZwatchingZexplicit_transactionrr/r/r1rW?szClusterPipeline.resetc CsR|sgStd|jD]0}z|j|||dWStk rBYqXqtddS)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. r)rallow_redirectionsz0CLUSTERDOWN error. Unable to rebuild the clusterN)r]r_send_cluster_commandsr)rrrrr4r/r/r1r_s z%ClusterPipeline.send_cluster_commandsc s^t|ddd}i}|D]}|jdd}|rD||sD||}n(|j|jd|i}|sltd|jdt|d krtd |j|d } | j } | |kr| | } t | |j} t | j | j| || <|| |q|} | D] }|q| D] }|q|D]}|j|jq td d |Dddd}|r|rtd|djdt|djjdt|dj|jd 7_|r|j|D]J}ztj |j|j|_Wn*t!k r}z ||_W5d}~XYnXqg}t|dddD]B}|jd |j"kr:|j"|jd |jf|j|_||jq|rZ|#||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. cSs|jSr9rr`r/r/r1rrz8ClusterPipeline._send_cluster_commands..)rr(NZ node_flagr-r.rMzToo many targets for command rcss |]}t|jtjr|VqdSr9)rrrrrCrr/r/r1 sz9ClusterPipeline._send_cluster_commands..cSs|jSr9rrr/r/r1rrz7An exception occurred during pipeline execution. args: z , error: rcSs|jSr9rrr/r/r1rr)$sortedr8rr)r,rr7rr^rrr5 NodeCommandsr;r4r_rFwritereadr9r3rr3r+rr}rUrrrr=rsr6rrr)rrrrattemptr[rr2r(rbr>r6r3Z node_commandsrdr5r?rur/r1rsz          0   z&ClusterPipeline._send_cluster_commandscCs|s tddS)rz4ASK & MOVED redirection not allowed in this pipelineNr)rrr/r/r1_fail_on_redirectsz!ClusterPipeline._fail_on_redirectcGs|jd|S)NEXISTS)r)r6)rr%r/r/r1exists$szClusterPipeline.existscCs tddS)rz method eval() is not implementedNrrr/r/r1eval'szClusterPipeline.evalcCs tddS)rz!method multi() is not implementedNrrr/r/r1multi+szClusterPipeline.multicOs tddS)rz5method immediate_execute_command() is not implementedNrrr/r/r1immediate_execute_command/sz)ClusterPipeline.immediate_execute_commandcOs tddS)rz0method _execute_transaction() is not implementedNrrr/r/r1_execute_transaction5sz$ClusterPipeline._execute_transactioncCs tddS)rz(method load_scripts() is not implementedNrrr/r/r1 load_scripts9szClusterPipeline.load_scriptscGs tddS)rz!method watch() is not implementedNrrnamesr/r/r1watch=szClusterPipeline.watchcCs tddS)rz#method unwatch() is not implementedNrrr/r/r1unwatchAszClusterPipeline.unwatchcOs tddS)rz4method script_load_for_pipeline() is not implementedNrrr/r/r1script_load_for_pipelineEsz(ClusterPipeline.script_load_for_pipelinecGs$t|dkrtd|d|dS)z7 "Delete a key specified by ``names``" rMz=deleting multiple keys is not implemented in pipeline commandZDELr)r^rr6rr/r/r1deleteKs  zClusterPipeline.delete)NNNFr7rN)T)TT)TT)&r}r~rrXrr"rrr#rrrMrrrr}r~rr6rrrrrWrrrrrrrrrrrrrryr/r/rur1rs\  %   ! &r.)rr-csfdd}|S)zi Prints error because some pipelined commands should be blocked when running in cluster-mode cstdddS)Nz"ERROR: Calling pipelined function z1 is blocked when running redis in cluster mode...r)r7rvrr/r1inner]s z%block_pipeline_command..innerr/)rrr/rr1block_pipeline_commandWs r)EZ BGREWRITEAOFrZBITOPZ BRPOPLPUSHrrrrZCLIENTrrrrZCONFIGrECHOr"rrrrrZMGETzMGET NONATOMICZMOVEZMSETzMSET NONATOMICZMSETNXZPFCOUNTZPFMERGErZPUBLISHrrrZRENAMEZRENAMENXZ RPOPLPUSHrrrrz SCRIPT KILLrZSCRIPTZSDIFFZ SDIFFSTOREz SENTINEL GET MASTER ADDR BY NAMEzSENTINEL MASTERzSENTINEL MASTERSzSENTINEL MONITORzSENTINEL REMOVEzSENTINEL SENTINELSz SENTINEL SETzSENTINEL SLAVESZSENTINELrZSINTERZ SINTERSTOREZSLAVEOFrrrZSLOWLOGZSMOVEZSORTZSUNIONZ SUNIONSTORErrr4c@seZdZdZdddZdS)rrNcCs4||_|dkri}||_||_d|_d|_d|_dS)NF)r7r8rrrbr@)rr7r8rr/r/r1rszPipelineCommand.__init__)NN)r}r~rrXrr/r/r/r1rsrc@s0eZdZdZddZddZddZdd Zd S) rrcCs||_||_||_g|_dSr|)r;r4r3commands)rr;r4r3r/r/r1rszNodeCommands.__init__cCs|j|dSr|)rr_)rrr/r/r1r_szNodeCommands.appendc Csv|j}|j}|D] }d|_qz||dd|DWn6ttfk rp}z|D] }||_qTW5d}~XYnXdS)z= Code borrowed from Redis so it can be fixed NcSsg|] }|jqSr/rNrr/r/r1rEsz&NodeCommands.write..)r3rrZsend_packed_commandZ pack_commandsrr")rr3rrr5r/r/r1rszNodeCommands.writec Cs|j}|jD]}|jdkr z|j||jdf|j|_Wq ttfk rx}z|jD] }||_qTWYdSd}~XYq tk rt d|_Yq Xq dS)rNrrM) r3rrr;r7r8rr"rsysexc_info)rr3rr5r/r/r1rs   zNodeCommands.readN)r}r~rrXrr_rrr/r/r/r1rs r)YrloggingrrJrrr< collectionsrtypingrrrrZ redis.clientrrr r Zredis.commandsr r r Zredis.connectionrrrrZ redis.crcrrZredis.exceptionsrrrrrrrrrrrrr r!r"r#Z redis.lockr%Z redis.utilsr&r'r(r)r* getLoggerr}rrUr8r2r5rBrHrZrcrrrrqrrrxryrrrrPrr rrZPIPELINE_BLOCKED_COMMANDSr;replacelowerrDrrr/r/r/r1st H      I1hG