o >h `@s,ddlZddlZddlZddlZddlZddlZddlZddlmZm Z ddl m Z ddl m Z ddl mZmZmZmZmZmZmZmZmZmZmZmZmZmZddlmZmZddlm Z m!Z!m"Z"ddl#m$Z$dd l%m&Z&m'Z'm(Z(dd l)m*Z*dd l+m,Z,dd l-m.Z.dd l/m0Z0m1Z1ddl2m3Z3m4Z4m5Z5ddl6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@ddlAmBZBmCZCddlDmEZEmFZFddlGmHZHddlImJZJmKZKddlLmMZMmNZNmOZOmPZPmQZQmRZRmSZSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\m]Z]m^Z^ddl_m`Z`maZambZbddlcmdZdmeZemfZfmgZgmhZhmiZimjZjedrddlkmlZlmmZmndZldZmedendedeedfZoGddde5e;eCZpGdddZqGdddZrGdd d e5e;eCZse7D]Ztetud!d"vZtetd#krcqRewesete>etqRGd$d%d%ZxGd&d'd'eZyGd(d)d)eyZzGd*d+d+ezZ{Gd,d-d-ezZ|dS).N)ABCabstractmethod)copy)chain)AnyCallable CoroutineDequeDict GeneratorListMappingOptionalSetTupleTypeTypeVarUnion)AsyncCommandsParserEncoder)_RedisCallbacks_RedisCallbacksRESP2_RedisCallbacksRESP3)ResponseCallbackT) Connection SSLConnection parse_urlLock)Retry)TokenInterface)ExponentialWithJitterBackoff NoBackoff)EMPTY_RESPONSE NEVER_DECODE AbstractRedis) PIPELINE_BLOCKED_COMMANDSPRIMARYREPLICASLOT_IDAbstractRedisCluster LoadBalancerLoadBalancingStrategyblock_pipeline_command get_node_nameparse_cluster_slots) READ_COMMANDSAsyncRedisClusterCommands)REDIS_CLUSTER_HASH_SLOTSkey_slot)CredentialProvider)#AfterAsyncClusterInstantiationEventEventDispatcher)AskErrorBusyLoadingErrorClusterDownError ClusterErrorConnectionErrorCrossSlotTransactionError DataErrorExecAbortErrorInvalidPipelineStackMaxConnectionsError MovedErrorRedisClusterException RedisError ResponseErrorSlotNotCoveredError TimeoutError TryAgainError WatchError)AnyKeyT EncodableTKeyT) SSL_AVAILABLEdeprecated_argsdeprecated_functionget_lib_versionsafe_str str_if_bytes truncate_text) TLSVersion VerifyMode TargetNodesT ClusterNodecV@seZdZdZedededdfddZdZe dgd d d e d gd dd ddddddddddddddddddde ddddddddddddddddddddf(de ede ee fde edd eded!e ed"ed#e d e d$e d%e d&d'e eeed(e ee fd)e ed*e ed+e ed,e ed-e ed.e ed/e ed0ed1ed2ed3ed4e ed5ed6e ee e e effd7e ed8ed9e ed:e ed;e eefde ed?e ed@e edAe e dBe eeee fgeee ffdCe eddfRdDdEZddFdGZddHdIZedJdKdLdMddNdOZ ddPdQZ!ddUdVZ"de#eddffdWdXZ$dYZ%e&j'e(j)fdZed[eddfd\d]Z*d^e+ddfd_d`Z,dedfdadbZ-dedfdcddZ.dedfdedfZ/ddgdhZ0ddidjZ1ddldmZ2   dde ede e dne ede dfdodpZ3 ddqedrede dfdsdtZ4dqe5de fdudvZ6de7fdwdxZ8de9ee effdydzZ:d%e;ddfd{d|Zddd}edede ededfddZ?d}edede fddZ@dedefddZAdededfddZBde5dedefddZCddde eDe5fdedefddZE dde ede eddfddZF      ddeDde ededede ede eeGdededeGfddZHdeIddeffddZJdS) RedisClustera Create a new RedisCluster client. Pass one of parameters: - `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 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 read_from_replicas: | @deprecated - please use load_balancing_strategy instead | Enable read from replicas in READONLY mode. When set to true, read commands will be assigned between the primary and its replications in a Round-Robin manner. The data read from replicas is eventually consistent with the data in primary nodes. :param load_balancing_strategy: | Enable read from replicas in READONLY mode and defines the load balancing strategy that will be used for cluster node selection. The data read from replicas is eventually consistent with the data in primary nodes. :param dynamic_startup_nodes: | Set the RedisCluster's startup nodes to all 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 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 cluster_error_retry_attempts: | @deprecated - Please configure the 'retry' object instead In case 'retry' object is set - this argument is ignored! Number of times to retry before raising an error when :class:`~.TimeoutError`, :class:`~.ConnectionError`, :class:`~.SlotNotCoveredError` or :class:`~.ClusterDownError` are encountered :param retry: | A retry object that defines the retry strategy and the number of retries for the cluster client. In current implementation for the cluster client (starting form redis-py version 6.0.0) the retry object is not yet fully utilized, instead it is used just to determine the number of retries for the cluster client. In the future releases the retry object will be used to handle the cluster client retries! :param max_connections: | Maximum number of connections per node. If there are no free connections & the maximum number of connections are already created, a :class:`~.MaxConnectionsError` is raised. :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. | Rest of the arguments will be passed to the :class:`~redis.asyncio.connection.Connection` instances when created :raises RedisClusterException: if any arguments are invalid or unknown. Eg: - `db` != 0 or None - `path` argument for unix socket connection - none of the `host`/`port` & `startup_nodes` were provided urlkwargsreturncKs4|t||ddturd|d<|di|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 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: 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. 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 are used. connection_classNTssl)updaterpopr)clsrXrYr]r]u/var/www/vedio/testing/chatpythonscript.ninositsolution.com/env/lib/python3.10/site-packages/redis/asyncio/cluster.pyfrom_urlszRedisCluster.from_url) _initialize_lockretry command_flagscommands_parserconnection_kwargsencoder node_flags nodes_managerread_from_replicasreinitialize_counterreinitialize_stepsresponse_callbacksresult_callbacksrlz6Please configure the 'load_balancing_strategy' insteadz5.3.0) args_to_warnreasonversioncluster_error_retry_attemptsz+Please configure the 'retry' object insteadz6.0.0NiTFrzredis-pyzutf-8strictrequiredhostport startup_nodesrVrequire_full_coverageload_balancing_strategydynamic_startup_nodesrnmax_connectionsrerretry_on_errordbpathcredential_providerusernamepassword client_namelib_name lib_versionencodingencoding_errorsdecode_responseshealth_check_intervalsocket_connect_timeoutsocket_keepalivesocket_keepalive_optionssocket_timeoutr\ ssl_ca_certs ssl_ca_data ssl_cert_reqs ssl_certfilessl_check_hostname ssl_keyfilessl_min_version ssl_ciphersprotocol address_remapevent_dispatcherc), CsF| rtd|r td|r|s|stdid| dtd|d|d|d |d |d |d |d |d|d|d|d|d|d|d|&})|r\|)t||| |!|"|#|$|%d |s`|re|j|)d<| rk| |_n ttddd| d|_| r~|j| t |)d<|) ddvr|)dt n|)dt |)|_|rg}*|D]}+|*t|+j|+jfi|jq|*}ng}|r|r|t||fi|j|(durt|_n|(|_t|||)||'|jd|_t||||_||_||_||_d|_t|_|jj |_!|jj" |_#|)d|_$|jj% |_&dd |j&d!<d"|_'d|_(dS)#Nz/Argument 'db' must be 0 or None in cluster modez3Unix domain socket is not supported in cluster modea1RedisCluster requires at least one node to discover the cluster. Please provide one of the following or use RedisCluster.from_url: - host and port: RedisCluster(host="localhost", port=6379) - startup_nodes: RedisCluster(startup_nodes=[ClusterNode("localhost", 6379), ClusterNode("localhost", 6380)])rr[rrrrrrrrrrrrrrr) r[rrrrrrrrredis_connect_func )basecap)backoffretriesro)3rv)rrrrc[stt|dfi|SNr)r/listvalues)cmdresrYr]r]ras z'RedisCluster.__init__.. CLUSTER SLOTST))rBrr^r on_connectrerr!update_supported_errorsrrgetrrrhappendrVr{r|r6_event_dispatcher NodesManagerrkrrirlrrnrmrrg __class__ NODE_FLAGSrj COMMAND_FLAGSrfroRESULT_CALLBACKSrprcrd),selfr{r|r}r~rlrrrnrtrrerrrrrrrrrrrrrrrrrr\rrrrrrrrrrrrY passed_nodesnoder]r]ra__init__s<            zRedisCluster.__init__c s|jrj|js t|_|j4IdHL|jrIz|jIdH|j|jjIdHd|_Wn$tyH|j IdH|j dIdHwWdIdH|SWdIdH|S1IdHsewY|S)zJGet all nodes from startup nodes & creates connections if not initialized.NFr}) rcrdasynciorrk initializerg default_node BaseExceptionacloserr]r]rars2      zRedisCluster.initializec s|jsN|js t|_|j4IdH.|js6d|_|jIdH|jdIdHWdIdHdSWdIdHdS1IdHsGwYdSdS)z.Close all connections & client if initialized.NTr})rcrdrrrkrrr]r]rars .zRedisCluster.aclosez5.0.0zUse aclose() insteadclose)rsrrnamecs|IdHdS)z.alias for aclose() for backwards compatibilityNrrr]r]rarszRedisCluster.closec|IdHSNrrr]r]ra __aenter__zRedisCluster.__aenter__exc_type exc_value tracebackc|IdHdSrrrrrrr]r]ra __aexit__zRedisCluster.__aexit__cC |Srr __await__rr]r]rar zRedisCluster.__await__zUnclosed RedisCluster client_warn_grlcCsft|dr/|js1||jd|t|dz||jd}||WdSty.YdSwdSdS)Nrc sourceclientmessage)hasattrrc _DEL_MESSAGEResourceWarningcall_exception_handler RuntimeError)rrrcontextr]r]ra__del__s  zRedisCluster.__del__ connectioncsB|IdH|dIdHt|IdHdkrtddS)NREADONLYOKzREADONLY command failed)r send_commandrQ read_responser;rrr]r]rars zRedisCluster.on_connectcCst|jjS)zGet all nodes of the cluster.)rrk nodes_cacherrr]r]ra get_nodesszRedisCluster.get_nodescC |jtS)z%Get the primary nodes of the cluster.)rkget_nodes_by_server_typer'rr]r]ra get_primaries zRedisCluster.get_primariescCr)z%Get the replica nodes of the cluster.)rkrr(rr]r]ra get_replicasrzRedisCluster.get_replicascCstt|jjS)z!Get a random node of the cluster.)randomchoicerrkrrrr]r]raget_random_nodeszRedisCluster.get_random_nodecCs|jjS)z#Get the default node of the client.)rkrrr]r]raget_default_node szRedisCluster.get_default_nodercCs&|r |j|jds td||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_noderr=rkr)rrr]r]raset_default_node s zRedisCluster.set_default_nodercCs|j|||S)z&Get node by (host, port) or node_name.)rkrrr{r|rr]r]rarszRedisCluster.get_nodekeyreplicacCs`||}|jj|}|std|d|r*t|jj|dkr$dSd}||Sd}||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.rzNrr)keyslotrk slots_cacherrElen)rrrslot slot_cachenode_idxr]r]raget_node_from_key!s zRedisCluster.get_node_from_keycCst|j|S)z Find the keyslot for a given key. See: https://redis.io/docs/manual/scaling/#redis-cluster-data-sharding )r3riencode)rrr]r]rar=szRedisCluster.keyslotcC|jS)z%Get the encoder object of the client.)rirr]r]ra get_encoderEzRedisCluster.get_encodercCr)zGGet the kwargs passed to :class:`~redis.asyncio.connection.Connection`.)rhrr]r]raget_connection_kwargsIrz"RedisCluster.get_connection_kwargscC ||_dSr)re)rrer]r]ra set_retryM zRedisCluster.set_retrycommandcallbackcCs||j|<dS)zSet a custom response callback.N)ro)rr r r]r]raset_response_callbackPsz"RedisCluster.set_response_callback) node_flagargsrcs|s |j|}||jvrQ||jjkr|jjgS||jjkr%|jt S||jj kr1|jt S||jj kr?t |jjS||jjkrQtt |jjgS|j|j|g|RIdH|joe|tv|tvrn|jgSdgSr)rfrrjr DEFAULT_NODErkr PRIMARIESrr'REPLICASr( ALL_NODESrrrRANDOMrrget_node_from_slot_determine_slotrlr0r)rr rrr]r]ra_determine_nodesTs.            zRedisCluster._determine_nodescsj|tkrt|dS|dvr;t|dkr&td|g|R|ddt|d}|s:tdt Sn"j j |g|RIdH}|s]|dvrVtdt Std|t|dkrj |dSfdd |D}t|dkrt|d | S) Nr)EVALEVALSHArzzInvalid args in command: r)FCALLFCALL_ROzNo way to dispatch this command to Redis Cluster. Missing key. You can execute the command by specifying target nodes. Command: csh|]}|qSr])r).0rrr]ra z/RedisCluster._determine_slot..z) - all keys must map to the same key slot)rfrr)intupperrrBr randranger2rgget_keysrr_)rr rkeysslotsr]rrarws:       zRedisCluster._determine_slot target_nodescCst|to ||jvSr) isinstancestrrj)rr%r]r]ra _is_node_flagszRedisCluster._is_node_flagcCsRt|tr |}|St|tr|g}|St|tr t|}|Stdt|)Nztarget_nodes type can be one of the following: node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES),ClusterNode, list, or dict. The passed type is )r&rrVdictr TypeErrortype)rr%nodesr]r]ra_parse_target_nodess   z RedisCluster._parse_target_nodesc sd}g}d}j}dd}|r$|s$|}d}d}d|}t|D]} jrJIdHt|dkrJ|d krJ z|sbj d|iIdH}|sbt dd t|dkrj |dgRiIdH} |jvrj|||dj| ifiWS| WSd d |D} tjfd d |DIdH} |jvrȈj||tt| | fiWStt| | WSty} z|dkrt| jjvr|d8}WYd} ~ q,| d} ~ wwdS)a Execute a raw command on the appropriate cluster node or target_nodes. It will retry the command as specified by the retries property of the :attr:`retry` & 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 rFr%NTrr!No targets were found to execute command oncSg|]}|jqSr]rrrr]r]ra z0RedisCluster.execute_command..c3s.|]}tj|gRiVqdSr)r create_task_execute_commandr2rrYrr]ra s z/RedisCluster.execute_command..)re get_retriesr_r(r-rangercrrrreplace_default_noderrBr6rprrgatherr)zip Exceptionr+rERRORS_ALLOW_RETRY)rrrYr r%target_nodes_specifiedretry_attemptspassed_targetsexecute_attempts_retr#rer]r7raexecute_commandst        "     zRedisCluster.execute_command target_nodec sd}}d}|j}|dkr|d8}zC|r&|j|d}|dIdHd}n$|rJ|j|IdH}|j||jo<|dtv|dtvrE|jnd}d}|j|i|IdHWSt t fy_t t fyx|jj |jd|IdHttfy|IdHtdIdHty} z)|jd7_|jr|j|jdkr|IdHd|_n| |j_d}WYd} ~ n7d} ~ wty} zt| j| jd}d}WYd} ~ nd} ~ wty||jd krtd IdHYnw|dkstd ) NFrrrASKING?Tr{r|rzg?zTTL exhausted.)RedisClusterRequestTTLrrGrrkrrlr0rr8r@r;rFr}r_rrr9rErsleeprArmrn_moved_exceptionr7r.r{r|rGr:) rrHrrYaskingmoved redirect_addrttlrrFr]r]rar6sj     IzRedisCluster._execute_command transaction shard_hintClusterPipelinecCs|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 mode)rBrU)rrSrTr]r]rapipelinegs  zRedisCluster.pipeline皙?rtimeoutrMblockingblocking_timeout lock_class thread_localraise_on_release_errorc Cs$|durt}|||||||||dS)a} 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. ``raise_on_release_error`` indicates whether to raise an exception when the lock is no longer owned when exiting the context manager. By default, this is True, meaning an exception will be raised. If False, the warning will be logged and the exception will be suppressed. 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)rXrMrYrZr\r]r) rrrXrMrYrZr[r\r]r]r]ralockvsDzRedisCluster.lockfuncc s|dd}|dd}|dd}|d|4IdHB} z)|r*|j|IdH||IdH}|IdH} |r<|n| WWdIdHSty^|dur\|dkr\t|Yqw1IdHsewYdS)z Convenience method for executing the callable `func` as a transaction while watching all keys specified in `watches`. The 'func' callable should expect a single argument which is a Pipeline object. rTNvalue_from_callableF watch_delayTr)r_rVwatchexecuterHtimerM) rr_watchesrYrTr`rapipe func_value exec_valuer]r]rarSs&      zRedisCluster.transaction)rZrWrZNrNrNrNrZN)rZrV)rrVrZNNNNF)NN)NrWTNNTT)K__name__ __module__ __qualname____doc__ classmethodr'rrb __slots__rMrOrrrr boolr,rr>r4floatr bytesrTrSrrr6rrrrNrrrr rrwarningswarnrget_running_looprrrrrrrrrrrrJrrrr rrr rr rrr(r-rGrKr6rVrr^rrSr]r]r]rarWcse           !#$% &'()*+, -./ ,               #3T  S    Q rWc@s&eZdZdZdZ d1deddedeeefde ed ed e ed e d dfd dZ d efddZ de d efddZdZejejfde de d dfddZd2ddZd efddZded dfddZded ed!e d e fd"d#Zd$e d!e d e fd%d&Zd'ed(d efd)d*Zd+efd,d-Zd.efd/d0Z dS)3rVz Create a new ClusterNode. Each ClusterNode manages multiple :class:`~redis.asyncio.connection.Connection` objects for the (host, port). ) _connections_freerdrr[rhr{rrr|ro server_typeNrw)rr[r{r|r{rr[rhrZcKs|dkr t|}||d<||d<||_||_t|||_||_||_||_||_ | di|_ g|_ t j|jd|_|j dd|_|jdurNt|_dSdS)N localhostr{r|ro)maxlenr)socket gethostbynamer{r|r.rr{rr[rhr_rory collectionsdequerzrrr6)rr{r|r{rr[rhr]r]rars$    zClusterNode.__init__c Cs&d|jd|jd|jd|jd S)Nz[host=z, port=z, name=z, server_type=])r{r|rr{rr]r]ra__repr__s zClusterNode.__repr__objcCst|to |j|jkSr)r&rVr)rrr]r]ra__eq__szClusterNode.__eq__zUnclosed ClusterNode objectrrc Csh|jD].}|jr1||jd|t|dz||jd}||WdSty0YdSwqdS)Nrrr)ry is_connectedrrrr)rrrrrr]r]rar!s   zClusterNode.__del__csDtjdd|jDddiIdH}tdd|Dd}|r |dS)Ncs|] }t|VqdSrrr5 disconnect)rrr]r]rar83  z)ClusterNode.disconnect..return_exceptionsTcss|] }t|tr|VqdSr)r&r>)rrr]r]rar89s)rr<rynext)rrEexcr]r]rar1s zClusterNode.disconnectcCszz|jWSty<t|j|jkr9ttdtfd}|j }||d<|j di|}|j ||YSt w)Nr)rrsupported_errorsrer])rzpopleft IndexErrorrryrrr"r;rhrr[rr@)rrerhrr]r]raacquire_connection=s      zClusterNode.acquire_connectionrcCs|j|dS)z8 Release connection back to free queue. N)rzrrr]r]rareleaseXszClusterNode.releaser rYcszt|vr|jddIdH}|tn|IdH}Wnty/t|vr.|tYSwt|vr9|t|dd||jvrO|j||fi|S|S)NT)disable_decodingr#)r$rr_rDr#ro)rrr rYresponser]r]raparse_response^s$      zClusterNode.parse_responserc s^|}||j|dIdHz|j||dfi|IdHW|j|S|j|w)NFr)rsend_packed_command pack_commandrrzr)rrrYrr]r]rarGxs zClusterNode.execute_commandcommandsPipelineCommandc s|}||dd|DdIdHd}|D]-}z|j||jdfi|jIdH|_WqtyG}z ||_d}WYd}~qd}~ww|j ||S)Ncss|]}|jVqdSrrrrr]r]rar8sz/ClusterNode.execute_pipeline..FrT) rr pack_commandsrrrYresultr>rzr)rrrrErrFr]r]raexecute_pipelines(   zClusterNode.execute_pipelinetokencst}jr:jjfddfddIdHjfddfddIdH|js|rJ|j|s.c |Sr_mockerrorrr]rar csSr)rr])rr]rarscrrrrrr]rarr)rrrzrrecall_with_retryr)rr tmp_queuer])rrrrare_auth_callbacks"        zClusterNode.re_auth_callbackrcdS)z Dummy functions, needs to be passed as error callback to retry object. :param error: :return: Nr]rrr]r]rarszClusterNode._mockrri)!rmrnrorprrrr'rrrrrrrrsrrrvrwrrxrrrrrrGr rr rrCrr]r]r]rarVsd     c@sLeZdZdZ   d,deddedeeefded e e e ee fge ee ffd e e d dfd d Z   d-de ede e de ed e dfddZ d.deedfdeedfded dfddZddZd/ddZ  d0de ded dfdd Zd!ed edfd"d#Zd/d$d%Zd1d'ed dfd(d)Zdede d e ee ffd*d+ZdS)2r) _dynamic_startup_nodesrNrrhrrread_load_balancerr~rr}rTNr}rVr~rhrrrrZcCsfdd|D|_||_||_||_d|_i|_i|_t|_||_ d|_ |dur.t |_ dS||_ dS)NcSsi|]}|j|qSr]r1r2r]r]ra z)NodesManager.__init__..) r}r~rhrrrrr+rrrNr6r)rr}r~rhrrrr]r]rars   zNodesManager.__init__r{r|rcCsF|r|r|dkr t|}|jt||dS|r|j|Std)Nr|rKzEget_node requires one of the following: 1. node name 2. host and port)r~rrrr.r=rr]r]rars  zNodesManager.get_nodeFoldnew remove_oldcCsx|rt|D]}||vrt||}q|D]\}}||vr5|||ur,qt||}|||<qdSr)rr#rr5r_ritems)rrrrrtaskrr]r]ra set_nodess  zNodesManager.set_nodescCrr)rN)r exceptionr]r]raupdate_moved_exceptionr z#NodesManager.update_moved_exceptioncCs|j}|j|j|jd}|r|jtkrt|_nt|j|jtfi|j}||j |j |i||j |j vrd|j |j d}t |_|j |j ||j |j |||j |j d<|j|krc||_n|g|j |j <d|_dS)NrKr)rNrr{r|r{r'rVrhrrrrslot_idr(rremover)rrFredirected_node old_primaryr]r]ra_update_moved_slots s.    z NodesManager._update_moved_slotsrrlc Cs|jr||dur|durtj}z0t|j|dkr;|r;|j|dj}|j|t|j||}|j||WS|j|dWSt t fyWt d|d|j dw)NTrrrz5" not covered by the cluster. "require_full_coverage=") rNrr, ROUND_ROBINrrrrget_server_indexrr*rEr~)rrrlr primary_namerr]r]rar1s&zNodesManager.get_node_from_slotr{csfdd|jDS)Ncsg|] }|jkr|qSr]r{r2rr]rar3Ms  z9NodesManager.get_nodes_by_server_type..)rr)rr{r]rrarLs z%NodesManager.get_nodes_by_server_typec sl|ji}i}g}d}d}d}t|jD]S}z)z|jt|j|j dd| dIdH}Wn t yAt dwd}WntyY} z| }WYd} ~ qd} ~ wwt|dkrx|dddsxt|jdkrx|j|ddd<|D]} tdt| D] } d d | | D| | <q| d} | d} | d kr|j} t| d}|| |\} }g}| t| |}|st| |tfi|j }|||j<||| d d}|D]1}|d} |d}|| |\} }| t| |}|st| |tfi|j }|||j<||qtt| dt| ddD];} | |vr%||| <q|| d}|j|jkrR||jd |jd| t|dkrRt dd|qqzd}ttD] } | |vrfd}nq[|rmnq|s{t dt|||s|jrt dt|dtd||_|j|j|dd|j r|j|j|jdd|!td|_"d|_#dS)NFrrz(Cluster mode is not enabled on this nodeTrrrzcSsg|]}t|qSr])rQ)rvalr]r]rar3rz+NodesManager.initialize..rvz vs z on slot: ruz6startup_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)$rresettupler}rrdispatchr5rrhrrGrDrBr>rr{r:rremap_host_portr.rVr'rrr(joinr2r'r~rrrrrrN)rtmp_nodes_cache tmp_slots disagreementsstartup_nodes_reachable fully_coveredr startup_node cluster_slotsrFri primary_noder{r|nodes_for_slotrH replica_nodes replica_nodetarget_replica_nodetmp_slotr]r]rarSs           "      zNodesManager.initializerattrcs0d|_tjddt||DIdHdS)Ncsrrrr2r]r]rar8rz&NodesManager.aclose..)rrr<getattrr)rrr]r]rars zNodesManager.aclosecCs|jr |||fS||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)rr{r|r]r]rarszNodesManager.remap_host_port)TNNrkrlri)FN)r)rmrnrorrr rsr r'rrrrrr6rrrrrrrrrrr]r]r]rarst        )   " rc@s4eZdZdZdZ d8dedeeddfddZd9d d Z d9d d Z d:ddZ de e ddffddZd9ddZd:ddZdefddZdefddZdeeefde ddfddZ d;d!ed"edee fd#d$Zd%ed&eddfd'd(Zd)d*Zd+d,Zd-d.Zd/d0Zd1d2Z d3d4Z!d5e"e#efddfd6d7Z$dS)rr]r]rar3s z-PipelineStrategy._execute..r%rr.r/rzToo many targets for command rcss(|]}t|d|dVqdS)rrN)rr5rr2r]r]rar8s  z,PipelineStrategy._execute..rrrrcSr0r])rrr]r]rar3r4) rYr_r(r-rrrBrrrrr<ranyr&rrGrAr7rGr>rrrPrrRrrr+rWr?r;)rrrrrtodor,rrBr%rerrorsrFrr rdefault_cluster_noderr]r]rars           zPipelineStrategy._executecs g|_dSr)rrr]r]rars zPipelineStrategy.resetcCtd)Nz@method multi() is not supported outside of transactional contextrBrr]r]rarszPipelineStrategy.multic td)Nz@method watch() is not supported outside of transactional contextr#rr]r]rarbzPipelineStrategy.watchcr$)NzBmethod unwatch() is not supported outside of transactional contextr#rr]r]rarr%zPipelineStrategy.unwatchcr$)NzBmethod discard() is not supported outside of transactional contextr#rr]r]rarr%zPipelineStrategy.discardcs&t|dkr td|d|dS)Nrz>unlinking multiple keys is not implemented in pipeline commandUNLINKr)rrBrGrr]r]rars  zPipelineStrategy.unlinkr)rmrnrorUrr rIrJrrsr rrcrrrrbrrr __classcell__r]r]rrarPsD   $ TrcsteZdZdhZddhZhdZeefZe e e e fZ deddffdd Zdeeeffd d Zd eeefd edd fddZd eeefd edeffdd ZddZddZddZdedefddZddZddZd e e!efdd!fd"d#Z" $d;d%e#d&e#de$efd'd(Z%d)e$d*d%e#fd+d,Z&d)e$d*d%e#fd-d.Z'd/d0Z(d1d2Z)d3d4Z*d5d6Z+d7d8Z,d9d:Z-Z.S)EXECDISCARDr(rfrZNcsZt|d|_d|_t|_d|_d|_d|_t |j j j |_ |j tj|jdS)NF)rr_explicit_transaction _watchingset_pipeline_slots_transaction_node_transaction_connection _executingrr rre_retryrrWr?SLOT_REDIRECT_ERRORSr rr]rars  zTransactionStrategy.__init__cCsR|jstd|jjjt|jdd}||_|js#|j }||_|j|jfS)ao Find a connection for a pipeline transaction. For running an atomic transaction, watch keys ensure that contents have not been altered as long as the watch commands for those keys were sent over the same connection. So once we start watching a key, we fetch a connection to the node that owns that slot and reuse it. z:At least a command with a key is needed to identify a noderF) r/rBr rrkrrr0r1r)rrrr]r]ra*_get_client_and_connection_for_transactions    z>TransactionStrategy._get_client_and_connection_for_transactionrrYrcsDddfdd}tj|d}||r S)Nc sHztjiWdSty#}z |WYd}~dSd}~wwr)rrunr6r>)rFrrrYrrr]rarunner3s z3TransactionStrategy.execute_command..runner)target) threadingThreadstartr)rrrYr8threadr]r7rarG.s z#TransactionStrategy.execute_commandcs|jjjr|jjIdHd}|d|jvr"|jjj|IdH}|js,|d|jvri|jsi|ddkr9| |durP|j rI||j vrIt d|j |n|d|jvrat d|dd|j|i|S|durs|j |tj|i|S)Nrr)z0Cannot watch or send commands on different slotsz)Cannot identify slot number for command: z(,it cannot be triggered in a transaction)r rrcrNO_SLOTS_COMMANDSrr-IMMEDIATE_EXECUTE_COMMANDSr,_validate_watchr/r<addrB_immediate_execute_commandrrG)rrrY slot_numberrr]rar6Ds6   z$TransactionStrategy._execute_commandcCs|jrtdd|_dS)N"Cannot issue a WATCH after a MULTIT)r,rCr-rr]r]rar@hs z#TransactionStrategy._validate_watchcs$jfddjIdHS)NcsjiSr) _get_connection_and_send_commandr]roptionsrr]rarpsz@TransactionStrategy._immediate_execute_command..r3r_reinitialize_on_error)rrrGr]rFrarBns  z.TransactionStrategy._immediate_execute_commandcs4|\}}|j|||dg|Ri|IdHSr)r5_send_command_parse_response)rrrG redis_noderr]r]rarEts   z4TransactionStrategy._get_connection_and_send_commandrrKcs@|j|IdH|j||fi|IdH}||jvrd|_|S)z7 Send a command and parse the response NF)rrUNWATCH_COMMANDSr-)rrrK command_namerrGoutputr]r]rarJzs   z0TransactionStrategy._send_command_parse_responsecs|jrt||jvr|jrtdt||jvs t||jvrV|jr&d|_|jjj d7_ |jjj rN|jjj |jjj dkrN|jjj IdHd|_ n|jjj |d|_dS)Nz-Slot rebalancing occurred while watching keysrrF)r-r+r4r2rHCONNECTION_ERRORSr1r rrmrnrkrrrr]r]rarIs( z*TransactionStrategy._reinitialize_on_errorcCs<t||D]\}}t|tr|||jd|j|qdS)z8 Raise the first exception on the stack rN)r=r&r>rrr)r responsesrrrr]r]ra_raise_first_errors  z&TransactionStrategy._raise_first_errorrrUcCr")Nz1Method is not supported in transactional context.)NotImplementedErrorrr]r]rarsz"TransactionStrategy.mset_nonatomicTrrcs.|j}|s|jr |jsgS|||IdHSr)rr-r/!_execute_transaction_with_retries)rrrrr]r]rarcs zTransactionStrategy.executerrcs$jfddjIdHS)Ncs Sr)_execute_transactionr]rrrr]rars zGTransactionStrategy._execute_transaction_with_retries..rH)rrrr]rVrarTs  z5TransactionStrategy._execute_transaction_with_retriesc s"t|jdkr tdd|_|\}}ttddg|tddg}dd|D}||}||IdHg}z | |dIdHWn2t yb}z| |dd| |WYd}~nd}~w|j yv} z| | ddd} ~ wwt|jD]x\} } t| jvr| | | jtfq|z | |d IdH} Wq||jy} z| | | d| j| | WYd} ~ q|d} ~ w|j y} z | | | d| jd} ~ wt y}z| || d| j| |WYd}~q|d}~wwd}z | |dIdH}Wnty|r|dwd |_d |_|dur"td |D] \} }|| |q$t|t|jkrJtd d d|jDt||sTt|dkr[|||jg}t||jD]+\}}t|ts|jd}||jjj vr|jjj ||fi|j}| |qc|S)NrzDAll keys involved in a cluster transaction must map to the same slotTrMULTIr*cSsg|] }t|jvr|jqSr])r#rYrrcr]r]rar3sz.rDFzWatched variable changed.zeUnexpected response length for cluster pipeline EXEC. Command stack was {} but response had length {}cSsg|]}|jdqS)rrrXr]r]rar3 r)!rr/r<r2r5rrrrrrDrrrO enumeraterr#rYr4rr>r-rHinsertr?formatrRr=r&r>r rro)rrrrKrrpacked_commandsr rF cluster_errorrr rD slot_errorrdatarQrrMr]r]rarUs          z(TransactionStrategy._execute_transactioncsg|_|jr:z|jdIdH|jIdH|j|jd|_Wn|jy9|jr7|jIdHYnwd|_d|_d|_ t |_ d|_ dS)Nr(F) rr1rrr0rrOrr-r,r.r/r2rr]r]rar* s$  zTransactionStrategy.resetcCs&|jrtd|jrtdd|_dS)Nz"Cannot issue nested calls to MULTIz:Commands without an initial WATCH have already been issuedT)r,rCrrr]r]rarE s zTransactionStrategy.multics(|jrtd|jdg|RIdHS)NrDr))r,rCrGrr]r]rarbN szTransactionStrategy.watchcs|jr |dIdHSdS)Nr(T)r-rGrr]r]rarT szTransactionStrategy.unwatchcrrrrr]r]rarZ rzTransactionStrategy.discardcs|jdg|RS)Nr&)rGrr]r]rar] rzTransactionStrategy.unlinkr)/rmrnror>r?rLr7rAr4r;OSErrorr9rErOrUrrrVrr5rrKrJrrGr6r@rBrErJrIrRr rIrrsr rcrTrUrrrbrrrr'r]r]rrarsz   $       d r)}rrrr~r:rdrvabcrrr itertoolsrtypingrrrr r r r r rrrrrrredis._parsersrrredis._parsers.helpersrrrredis.asyncio.clientrredis.asyncio.connectionrrrredis.asyncio.lockrredis.asyncio.retryrredis.auth.tokenr redis.backoffr!r" redis.clientr#r$r% redis.clusterr&r'r(r)r*r+r,r-r.r/redis.commandsr0r1 redis.crcr2r3redis.credentialsr4 redis.eventr5r6redis.exceptionsr7r8r9r:r;r<r=r>r?r@rArBrCrDrErFrGrH redis.typingrIrJrK redis.utilsrLrMrNrOrPrQrRr\rSrTr'rUrWrVrrUr replacelowersetattrrrr rrr]r]r]rasz  @    0  P$ \8#  dF+