U d@s(ddlZddlZddlZddlZddlZddlmZmZmZm Z m Z m Z m Z m Z mZmZddlmZddlmZmZmZmZddlmZddlmZmZmZddlmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&ddl'm(Z(m)Z)dd l*m+Z+m,Z,dd l-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;dd lZ>m?Z?dd l@mAZAmBZBmCZCed eDde deedfZEdZFeeeDefdddZGGdddeZHGdddee"e)ZIGdddZJGdddZKGdddee"e)ZLeD]4ZMeMNddOZMeMdkrqePeLeMe$eMqGdd d ZQdS)!N) AnyDequeDict GeneratorListMappingOptionalTypeTypeVarUnion)ResponseCallbackT) Connection DefaultParserEncoder parse_url)CommandsParser)EMPTY_RESPONSE NEVER_DECODE AbstractRedis) PIPELINE_BLOCKED_COMMANDSPRIMARYREPLICASLOT_IDAbstractRedisCluster LoadBalancerblock_pipeline_command get_node_nameparse_cluster_slots) READ_COMMANDSAsyncRedisClusterCommands)REDIS_CLUSTER_HASH_SLOTSkey_slot)AskErrorBusyLoadingErrorClusterCrossSlotErrorClusterDownError ClusterErrorConnectionError DataErrorMasterDownError MovedErrorRedisClusterException ResponseErrorSlotNotCoveredError TimeoutError TryAgainError)AnyKeyT EncodableTKeyT) dict_mergesafe_str str_if_bytes TargetNodesT ClusterNode)Z client_namedbdecode_responsesZ encoder_classencodingencoding_errorsZhealth_check_interval parser_classpasswordredis_connect_funcretryZretry_on_timeoutZsocket_connect_timeoutZsocket_keepaliveZsocket_keepalive_optionsZsocket_read_sizesocket_timeout socket_typeusername)kwargsreturncKsdd|DS)z0Remove unsupported or disabled keys from kwargs.cSsi|]\}}|tkr||qS)CONNECTION_ALLOWED_KEYS).0kvrErE9/tmp/pip-unpacked-wheel-f5h5_hbx/redis/asyncio/cluster.py Tsz"cleanup_kwargs..)items)rCrErErJcleanup_kwargsRsrMc @s&eZdZeejeeee e e dZdS) ClusterParser)ZASKZTRYAGAINMOVED CLUSTERDOWNZ CROSSSLOT MASTERDOWNN) __name__ __module__ __qualname__r3rZEXCEPTION_CLASSESr"r/r*r%r$r)rErErErJrNWsrNc @s`eZdZdZeeeddddZdZdTe ee e e d e e e e e eedd d dZ ddddZddddZddddZdddddddZeeddfdddZdZddddZedddd Ze d dd!d"Ze d dd#d$Ze d dd%d&Zd dd'd(Zd dd)d*Zd dd+d,d-ZdUe ee e e ee d d.d/d0ZdVee e d d1d2d3Ze e d4d5d6Z!e"dd7d8Z#e$ee efdd9d:Z%ee&dd;deee ee d d?d@dAZ(eee dBdCdDZ)ee dEdFdGZ*ee d dEdHdIZ+e eedJdKdLZ,d e-e.e feedMdNdOZ/dWe ee edPdQdRdSZ0dS)X RedisClustera Create a new RedisCluster client. Pass one of parameters: - `url` - `host` & `port` - `startup_nodes` | Use ``await`` :meth:`initialize` to find cluster nodes & create connections. | Use ``await`` :meth:`close` to disconnect connections & close client. Many commands support the target_nodes kwarg. It can be one of the :attr:`NODE_FLAGS`: - :attr:`PRIMARIES` - :attr:`REPLICAS` - :attr:`ALL_NODES` - :attr:`RANDOM` - :attr:`DEFAULT_NODE` Note: This client is not thread/process/fork safe. :param host: | Can be used to point to a startup node :param port: | Port used if **host** is provided :param startup_nodes: | :class:`~.ClusterNode` to used as a startup node :param cluster_error_retry_attempts: | Retry command execution attempts when encountering :class:`~.ClusterDownError` or :class:`~.ConnectionError` :param require_full_coverage: | When set to ``False``: 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 :class:`~.ClusterDownError` for some key-based commands. | When set to ``True``: all slots must be covered to construct the cluster client. If not all slots are covered, :class:`~.RedisClusterException` will be thrown. | See: https://redis.io/docs/manual/scaling/#redis-cluster-configuration-parameters :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 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 url: | See :meth:`.from_url` :param kwargs: | Extra arguments that will be passed to the :class:`~redis.asyncio.connection.Connection` instances when created :raises RedisClusterException: if any arguments are invalid. Eg: - db kwarg - db != 0 in url - unix socket connection - none of host & url & startup_nodes were provided )urlrCrDcKs|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 :class:`~redis.asyncio.connection.Connection` when created. In the case of conflicting arguments, querystring arguments always win. rVrE)clsrVrCrErErJfrom_urls*zRedisCluster.from_url) _initialize_lockcluster_error_retry_attempts command_flagscommands_parserconnection_kwargsencoder node_flags nodes_managerread_from_replicasreinitialize_counterreinitialize_stepsresponse_callbacksresult_callbacksNF r7) hostport startup_nodesrequire_full_coveragerbr[rdrVrCrDc Ks~|sg}d| krtd|rrt|} d| kr4tdd| krP| ddkrPtd| | | d}| d|}n|rz|s|std |j| d <tf| |_} |jj |_ | d <|r|r| t ||f|jt f||d |j|_t| d d| dd| dd|_||_||_||_d|_t|_|jj |_|jj |_|jj |_dd|jd<d|_t|_ dS)Nr8z4Argument 'db' is not possible to use in cluster modepathzFRedisCluster does not currently support Unix Domain Socket connectionsrz9A ``db`` querystring option can only be 0 in cluster moderjrka5RedisCluster 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>re)rlrmr:zutf-8r;strictr9Fc[stt|df|S)Nr)rlistvalues)cmdresrCrErErJ:sz'RedisCluster.__init__.. CLUSTER SLOTST)!r+rupdateget on_connectrMr^ __class__RESPONSE_CALLBACKScopyreappendr7 NodesManagerrarr_r[rbrdrcrr]Z NODE_FLAGSr`Z COMMAND_FLAGSr\ZRESULT_CALLBACKSrfrYasyncioLockrZ) selfrjrkrlrmrbr[rdrVrCZ url_optionsrErErJ__init__sp         zRedisCluster.__init__rDc s|jr|j4IdHv|jrz0|jIdH|j|jjIdHd|_Wn8tk r|jIdH|jdIdHYnXW5QIdHRX|S)zJGet all nodes from startup nodes & creates connections if not initialized.NFrl)rYrZra initializer] default_node BaseExceptioncloserrErErJr@s  zRedisCluster.initializec sX|jsT|j4IdH4|jsDd|_|jIdH|jdIdHW5QIdHRXdS)z.Close all connections & client if initialized.NTrl)rYrZrarrrErErJrQs zRedisCluster.closecs|IdHSNrrrErErJ __aenter__ZszRedisCluster.__aenter__exc_type exc_value tracebackrDcs|IdHdSr)rrrrrrErErJ __aexit__]szRedisCluster.__aexit__cCs |Srr __await__rrErErJr`szRedisCluster.__await__zUnclosed RedisCluster clientcCsdt|dr`|js`tj|jd|t|dz||jd}t|Wnt k r^YnXdS)NrY sourceclientmessage) hasattrrYwarningswarn _DEL_MESSAGEResourceWarningr~get_event_loopcall_exception_handler RuntimeError)rcontextrErErJ__del__es zRedisCluster.__del__) connectionrDcsP|t|IdH|jrL|dIdHt|IdHdkrLtddS)NZREADONLYOKzREADONLY command failed) set_parserrNrxrbZ send_commandr5read_response_without_lockr')rrrErErJrxos  zRedisCluster.on_connectcCst|jjS)zGet all nodes of the cluster.)rpra nodes_cacherqrrErErJ get_nodes}szRedisCluster.get_nodescCs |jtS)z%Get the primary nodes of the cluster.)raget_nodes_by_server_typerrrErErJ get_primariesszRedisCluster.get_primariescCs |jtS)z%Get the replica nodes of the cluster.)rarrrrErErJ get_replicasszRedisCluster.get_replicascCstt|jjS)z!Get a random node of the cluster.)randomchoicerprarrqrrErErJget_random_nodeszRedisCluster.get_random_nodecCs|jjS)z#Get the default node of the client.)rarrrErErJget_default_nodeszRedisCluster.get_default_node)noderDcCs&|r|j|jdstd||j_dS)z Set the default node of the client. :raises DataError: if None is passed or node does not exist in cluster.  node_namez1The requested node does not exist in the cluster.N)get_nodenamer(rar)rrrErErJset_default_nodeszRedisCluster.set_default_noderjrkrrDcCs|j|||S)z&Get node by (host, port) or node_name.)rarrrjrkrrErErJrszRedisCluster.get_node)keyreplicarDcCs^||}|jj|}|s,td|d|rHt|jj|dkrHdS|rRd}nd}||S)aG Get the cluster node corresponding to the provided key. :param key: :param replica: | Indicates if a replica should be returned | None will returned if no replica holds this key :raises SlotNotCoveredError: if the key is not covered by any slot. Slot "z " is not covered by the cluster.Nr)keyslotra slots_cacherwr-len)rrrslotZ slot_cachenode_idxrErErJget_node_from_keys zRedisCluster.get_node_from_key)rrDcCst|j|S)z Find the keyslot for a given key. See: https://redis.io/docs/manual/scaling/#redis-cluster-data-sharding )r!r_encode)rrrErErJrszRedisCluster.keyslotcCs|jS)z%Get the encoder object of the client.)r_rrErErJ get_encoderszRedisCluster.get_encodercCs|jS)zGGet the kwargs passed to :class:`~redis.asyncio.connection.Connection`.)r^rrErErJget_connection_kwargssz"RedisCluster.get_connection_kwargs)commandcallbackrDcCs||j|<dS)zSet a custom response callback.N)re)rrrrErErJset_response_callbacksz"RedisCluster.set_response_callback) node_flag)rargsrrDcs|s|j|}||jkr||jjkr0|jjgS||jjkrH|jt S||jj kr`|jt S||jj kr|t |jjS||jjkrtt |jjgS|j|j|f|IdH|jo|tkgSr)r\rwr`ryZ DEFAULT_NODErarZ PRIMARIESrrZREPLICASrZ ALL_NODESrprrqZRANDOMrrget_node_from_slot_determine_slotrbr)rrrrrErErJ_determine_nodess$           zRedisCluster._determine_nodes)rrrDcsj|tkrt|dS|dkrjt|dkrDtd|f||dd|d}|stdtSn>j j |f|IdH}|s|dkrtdtStd|t|dkrˆ |dSfdd |D}t|dkrt|d | S) Nr)ZEVALZEVALSHArzInvalid args in command: r)ZFCALLZFCALL_ROzNo way to dispatch this command to Redis Cluster. Missing key. You can execute the command by specifying target nodes. Command: csh|]}|qSrE)r)rGrrrErJ sz/RedisCluster._determine_slot..z) - all keys must map to the same key slot) r\rwrintrr+r randranger r]Zget_keysrpop)rrrkeysslotsrErrJrs2     zRedisCluster._determine_slot) target_nodesrDcCst|to||jkSr) isinstancestrr`)rrrErErJ _is_node_flag(szRedisCluster._is_node_flagcCsPt|tr|}n, or dict. The passed type is )rrpr7dictrq TypeErrortype)rrnodesrErErJ_parse_target_nodes+s    z RedisCluster._parse_target_nodesrrCrDc sd}g}d}j}dd}|rB|sB|}d}d}t|D]b}jrdIdHz|sjd|iIdH}|stdd t |dkrj |dfIdH} |j krj |||dj | ifWS| WSd d |D} t jfd d |DIdH} |j krVj ||tt| | fWStt| | WSWqJtk r} z t| jjkr| } n| W5d} ~ XYqJXqJ| dS)a| Execute a raw command on the appropriate cluster node or target_nodes. It will retry the command as specified by :attr:`cluster_error_retry_attempts` & then raise an exception. :param args: | Raw command args :param kwargs: - target_nodes: :attr:`NODE_FLAGS` or :class:`~.ClusterNode` or List[:class:`~.ClusterNode`] or Dict[Any, :class:`~.ClusterNode`] - Rest of the kwargs are passed to the Redis connection :raises RedisClusterException: if target_nodes is not provided & the command can't be mapped to a slot rFrNTrr!No targets were found to execute command oncSsg|] }|jqSrErrGrrErErJ ssz0RedisCluster.execute_command..c3s&|]}tj|fVqdSr)r~ ensure_future_execute_commandrrrCrrErJ usz/RedisCluster.execute_command..)r[rrrrangerYrrr+rrrfrr~gatherrziprrryERRORS_ALLOW_RETRY)rrrCrrZtarget_nodes_specifiedZretry_attemptspassed_targets_retrrqe exceptionrErrJexecute_command?sh             zRedisCluster.execute_command) target_noderrCrDc sd}}d}|j}d}|dkr|d8}zn|rP|j|d}|dIdHd}n4|r|j|IdH} |j| |jo||dtk}d}|j||IdHWStk rYqt t fk r|d7}|dkrt dIdHn| IdHYqtk r`} zN|jd7_|jrD|j|jdkrD| IdHd|_n| |j_d}W5d} ~ XYqtk r||jd krt d IdHYqtk r} zt| j| jd }d}W5d} ~ XYqtk rt dIdH| IdHYqXqtd dS) NFrrrZASKING?Trg?rjrkzTTL exhausted.)ZRedisClusterRequestTTLrrrrarrbrr#r'r.r~sleeprr*rcrd_moved_exceptionr/r"rrjrkr%r&) rrrrCZaskingmovedZ redirect_addrZttlZconnection_error_retry_counterrrrErErJrs`    zRedisCluster._execute_commandClusterPipeline) transaction shard_hintrDcCs |r td|rtdt|S)z Create & return a new :class:`~.ClusterPipeline` object. Cluster implementation of pipeline does not support transaction or shard_hint. :raises RedisClusterException: if transaction or shard_hint are truthy values z(shard_hint is deprecated in cluster modez)transaction is deprecated in cluster mode)r+r)rrrrErErJpipelines zRedisCluster.pipeline)NrgNFFrhriN)NNN)F)NN)1rRrSrT__doc__ classmethodrrrX __slots__rrrboolrrrrrrrrrr rxrrrrrrrrr1rrrrrr rrrrrrr r2rrrErErErJrUesF+  W      3P  MrUc @seZdZdZdZddefeeeeee ee ddddZ edd d Z e e d d d ZdZddddZddddZedddZeee e dddZe e e dddZede dddZdS)r7z Create a new ClusterNode. Each ClusterNode manages multiple :class:`~redis.asyncio.connection.Connection` objects for the (host, port). ) _connections_freeconnection_classr^rjmax_connectionsrrkre server_typeNl)rjrkrrrr^rDcKs||dkrt|}||d<||d<||_||_t|||_||_||_||_||_ | dt j |_ g|_tj|jd|_dS)N localhostrjrkre)maxlen)socket gethostbynamerjrkrrrrrr^rrUrzrer collectionsdequer)rrjrkrrrr^rErErJrs"   zClusterNode.__init__rc Cs&d|jd|jd|jd|jd S)Nz[host=z, port=z, name=z, server_type=])rjrkrrrrErErJ__repr__s$zClusterNode.__repr__)objrDcCst|to|j|jkSr)rr7r)rr rErErJ__eq__%szClusterNode.__eq__zUnclosed ClusterNode objectc Csj|jD]^}|jrtj|jd|t|dz||jd}t|Wnt k r^YnXqfqdS)Nrrr) r is_connectedrrrrr~rrr)rrrrErErJr*s  zClusterNode.__del__csBtjdd|jDddiIdH}tdd|Dd}|r>|dS)Ncss|]}t|VqdSrr~r disconnect)rGrrErErJr:sz)ClusterNode.disconnect..Zreturn_exceptionsTcss|]}t|tr|VqdSr)r Exception)rGrsrErErJr@s )r~rrnext)rrexcrErErJr8s zClusterNode.disconnectcCs|jrHtt|jD](}|j}|jr0|S|j|q|jSt|j|jkrv|jf|j }|j||St ddS)NzToo many connections) rrrpopleftrr|rrrr^r')rrrrErErJacquire_connectionDs   zClusterNode.acquire_connection)rrrCrDcszz.t|kr|jddIdH}n|IdH}Wn*tk rXt|krR|tYSYnX||jkrv|j||f|S|S)NT)Zdisable_decoding)rrr,rre)rrrrCresponserErErJparse_responseUs  zClusterNode.parse_responserc sR|}||j|dIdHz|j||df|IdHWS|j|XdS)NFr)rsend_packed_commandZ pack_commandrr|r)rrrCrrErErJrjs zClusterNode.execute_commandPipelineCommand)commandsrDc s|}||dd|DdIdHd}|D]V}z$|j||jdf|jIdH|_Wq2tk r}z||_d}W5d}~XYq2Xq2|j ||S)Ncss|] }|jVqdSr)rrGrrrErErJr~sz/ClusterNode.execute_pipeline..FrT) rrZ pack_commandsrrrCresultrrr|)rrrrrrrrErErJexecute_pipelinexs&  zClusterNode.execute_pipeline)rRrSrTrrr rrrr rrr rrrrrrrrrrrErErErJr7s6   c@seZdZdZdedeeddddZdee ee ee eddd d Z de e dfe e dfedd d d Z ddddZd e eddddZe eddddZddddZd!e ddddZdS)"r})rr^rrread_load_balancerrmrrlFr7N)rlrmrCrDcKs@dd|D|_||_||_d|_i|_i|_t|_d|_dS)NcSsi|] }|j|qSrErrrErErJrKsz)NodesManager.__init__..) rlrmr^rrrrrr)rrlrmrCrErErJrszNodesManager.__init__rcCsJ|r.|r.|dkrt|}|jt||dS|r>|j|StddS)NrrzEget_node requires one of the following: 1. node name 2. host and port)rrrrwrr(rrErErJrs  zNodesManager.get_node)oldnew remove_oldrDcCsx|r2t|D] }||krt||q|D]8\}}||krj|||krXq:t|||||<q:dSr)rprr~rrrrL)rr r!r"rrrErErJ set_nodess zNodesManager.set_nodesrcCs|j}|j|j|jd}|r.|jtkrXt|_n*t|j|jtf|j}||j |j |i||j |j kr|j |j d}t |_|j |j ||j |j |||j |j d<|j|kr||_n|g|j |j <d|_dS)Nrr)rrrjrkrrr7r^r#rrrZslot_idrr|remover)rrZredirected_nodeZ old_primaryrErErJ_update_moved_slotss.  z NodesManager._update_moved_slots)rrbrDc Cs|jr|zL|rL|j|dj}|j|t|j|}|j||WS|j|dWSttfk rt d|d|j dYnXdS)Nrrz5" not covered by the cluster. "require_full_coverage=") rr%rrrZget_server_indexr IndexErrorrr-rm)rrrbZ primary_namerrErErJrs zNodesManager.get_node_from_slot)rrDcsfdd|jDS)Ncsg|]}|jkr|qSrErrr(rErJrs z9NodesManager.get_nodes_by_server_type..)rrq)rrrEr(rJrs z%NodesManager.get_nodes_by_server_typec s|ji}i}g}d}d}|jD]}z6|dIdHdsNtd|dIdH}d}Wnttfk rYq(Ynt k r}z:| } d| ksd| krWYq(ntd |d | W5d}~XYnBt k r}z"| } td |j d | W5d}~XYnXt |d krZ|d dd sZt |jd krZ|j|d dd <|D]tdt D]} dd| D| <qrd} | d } | dkr|j} t| d } |t| | }|st| | tf|j}|||j <ttd td d D]} | |krg|| <|| |fddtdt D}|D]V}|d } |d } |t| | }|st| | tf|j}|| ||||j <qPnZ|| d }|j |j kr ||j d|j d| t |dkr tdd|q q^d}ttD]} | |krd}q4q|r(q@q(|sNtd|sv|jrvtdt |dtd||_|j|j|dd|j|j|jdd|td |_d|_dS)NFINFOZcluster_enabledz(Cluster mode is not enabled on this noderuTrPrQz7ERROR sending "cluster slots" command to redis server: z . error: z6ERROR sending "cluster slots" command to redis server rrrcSsg|] }t|qSrE)r5)rGvalrErErJrLsz+NodesManager.initialize..csg|] }|qSrErE)rGjrrErJr_srhz vs z on slot: rz6startup_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") rresetrlrqrrwr+r'r.r,__str__rrrrjrrrr7rr^r|rjoinr rmrr#rrrr)rZtmp_nodes_cacheZ tmp_slotsZ disagreementsZstartup_nodes_reachableZ fully_coveredZ startup_nodeZ cluster_slotsrriZ primary_noderjrkrZ replica_nodesZ replica_nodeZtarget_replica_nodeZtmp_slotrEr-rJrs        "      zNodesManager.initializer)attrrDcs.d|_tjddt||DIdHdS)Ncss|]}t|VqdSrrrrErErJrsz%NodesManager.close..)rr~rgetattrrq)rr2rErErJrs  zNodesManager.close)F)NNN)F)F)r)rRrSrTrrrrrrrrrrr#r%rrrrrErErErJr}sF     ' r}c@s&eZdZdZdZeddddZdddd Zddd d Zddddd d dZ e e ddfdddZ ddddZ ddddd ddZedddZedddZeeefe ddddZd+eeee dddZd,d ed!eeee d"d#d$Zeedd%d&d'Zeeefdd(d)d*ZdS)-ra Create a new ClusterPipeline object. Usage:: result = await ( rc.pipeline() .set("A", 1) .get("A") .hset("K", "F", "V") .hgetall("K") .mset_nonatomic({"A": 2, "B": 3}) .get("A") .get("B") .delete("A", "B", "K") .execute() ) # result = [True, "1", 1, {"F": "V"}, True, True, "2", "3", 1, 1, 1] Note: For commands `DELETE`, `EXISTS`, `TOUCH`, `UNLINK`, `mset_nonatomic`, which are split across multiple nodes, you'll get multiple results for them in the array. Retryable errors: - :class:`~.ClusterDownError` - :class:`~.ConnectionError` - :class:`~.TimeoutError` Redirection errors: - :class:`~.TryAgainError` - :class:`~.MovedError` - :class:`~.AskError` :param client: | Existing :class:`~.RedisCluster` client )_command_stack_clientN)rrDcCs||_g|_dSr)r5r4)rrrErErJrszClusterPipeline.__init__rcs"|jjr|jIdHg|_|Sr)r5rYrr4rrErErJrszClusterPipeline.initializecs|IdHSrrrrErErJrszClusterPipeline.__aenter__rcs g|_dSrr4rrErErJrszClusterPipeline.__aexit__cCs |SrrrrErErJrszClusterPipeline.__await__cCs g|_|Srr6rrErErJ __enter__szClusterPipeline.__enter__cCs g|_dSrr6rrErErJ__exit__szClusterPipeline.__exit__cCs t|jSr)rr4rrErErJ__bool__szClusterPipeline.__bool__cCs t|jSr)rr4rrErErJ__len__szClusterPipeline.__len__rcOs"|jtt|jf|||S)ad Append a raw command to the pipeline. :param args: | Raw command args :param kwargs: - target_nodes: :attr:`NODE_FLAGS` or :class:`~.ClusterNode` or List[:class:`~.ClusterNode`] or Dict[Any, :class:`~.ClusterNode`] - Rest of the kwargs are passed to the Redis connection )r4r|rr)rrrCrErErJrszClusterPipeline.execute_commandT)raise_on_errorallow_redirectionsrDc s|js gSzt|jjD]}|jjr4|jIdHz&|j|j|j||dIdHWWhStk r}z>t||j j kr|}|j IdHt dIdHn|W5d}~XYqXq|W5g|_XdS)a  Execute the pipeline. It will retry the commands as specified by :attr:`cluster_error_retry_attempts` & then raise an exception. :param raise_on_error: | Raise the first error if there are any errors :param allow_redirections: | Whether to retry each failed command individually in case of redirection errors :raises RedisClusterException: if target_nodes is not provided & the command can't be mapped to a slot N)r;r<r)r4rr5r[rYr_executerrryrrr~r)rr;r<rrrrErErJexecute s*zClusterPipeline.executerUr)rstackr;r<rDc sdd|D}i}|D]}|jdd}|rB||sB||} n.|j|jd|iIdH} | sptd|jdt| dkrtd|j| d } | j|kr| gf|| j<|| jd |qt j d d | DIdH} t | r|rT|D]^}t|jtttfrz|j|j|jIdH|_Wqtk rP} z | |_W5d} ~ XYqXq|r|D]b}|j} t| tr^d tt|j}d |jdd|d| j}|f| jdd| _| q^dd|DS)NcSs"g|]}|jrt|jtr|qSrE)rrrrrErErJr@s z,ClusterPipeline._execute..rrrrrzToo many targets for command rcss&|]}t|d|dVqdS)rrN)r~rrrrErErJrZsz+ClusterPipeline._execute..rz Command #  (z) of pipeline caused error: cSsg|] }|jqSrE)rrrErErJrxs)rCrrrrrr+rrr|r~rrqanyrrr/r*r"rrr0mapr4position)rrr?r;r<todorrrrrrerrorsrrrmsgrErErJr=9sb        zClusterPipeline._execute)rrrDcGs*|j|D]}|j|f|q|Sr)r5Z_partition_keys_by_slotrqr)rrrZ slot_keysrErErJ_split_command_across_slotszsz+ClusterPipeline._split_command_across_slots)mappingrDcCs^|jj}i}|D](}t||d}||g|q|D]}|jd|qF|S)NrMSET)rI) r5r_rLr!r setdefaultextendrqr)rrHr_Z slots_pairspairrpairsrErErJmset_nonatomics  zClusterPipeline.mset_nonatomic)TT)TT)rRrSrTrrrUrrrrrrrr7r8rr9rr:r r2r1rrr>r=rrGrr0rNrErErErJrsL$   4 B rrrrNc@s.eZdZeeeddddZedddZdS)rN)rCrrCrDcOs||_||_||_d|_dSr)rrCrCr)rrCrrCrErErJrszPipelineCommand.__init__rcCsd|jd|jd|jdS)N[z] r@))rCrrCrrErErJr szPipelineCommand.__repr__)rRrSrTrrrrr rErErErJrsr)Rr~r rrrtypingrrrrrrrr r r Zredis.asyncio.clientr Zredis.asyncio.connectionr rrrZredis.asyncio.parserrZ redis.clientrrrZ redis.clusterrrrrrrrrrZredis.commandsrrZ redis.crcr r!Zredis.exceptionsr"r#r$r%r&r'r(r)r*r+r,r-r.r/Z redis.typingr0r1r2Z redis.utilsr3r4r5rr6rFrMrNrUr7r}rrreplacelowersetattrrrErErErJsX0  , @  &e