class ClusterCommandMixin:

    NODES_FLAGS = dict_merge({
        'CLUSTER INFO': NodeFlag.ALL_NODES,
        'CLUSTER COUNTKEYSINSLOT': NodeFlag.SLOT_ID
    },
    list_keys_to_dict(
        ['CLUSTER NODES', 'CLUSTER SLOTS'], NodeFlag.RANDOM
    )
    )

    RESPONSE_CALLBACKS = {
        'CLUSTER ADDSLOTS': bool_ok,
        'CLUSTER COUNT-FAILURE-REPORTS': lambda x: int(x),
        'CLUSTER COUNTKEYSINSLOT': lambda x: int(x),
        'CLUSTER DELSLOTS': bool_ok,
        'CLUSTER FAILOVER': bool_ok,
        'CLUSTER FORGET': bool_ok,
        'CLUSTER INFO': parse_cluster_info,
        'CLUSTER KEYSLOT': lambda x: int(x),
        'CLUSTER MEET': bool_ok,
        'CLUSTER NODES': parse_cluster_nodes,
        'CLUSTER REPLICATE': bool_ok,
        'CLUSTER RESET': bool_ok,
        'CLUSTER SAVECONFIG': bool_ok,
        'CLUSTER SET-CONFIG-EPOCH': bool_ok,
        'CLUSTER SETSLOT': bool_ok,
        'CLUSTER SLAVES': parse_cluster_nodes,
        'CLUSTER SLOTS': parse_cluster_slots,
        'ASKING': bool_ok,
        'READONLY': bool_ok,
        'READWRITE': bool_ok,
    }

    RESULT_CALLBACKS = dict_merge(
        list_keys_to_dict(
            ['CLUSTER INFO', 'CLUSTER ADDSLOTS', 'CLUSTER COUNT-FAILURE-REPORTS',
            'CLUSTER DELSLOTS', 'CLUSTER FAILOVER', 'CLUSTER FORGET'], lambda res: res
        )
    )

    def _nodes_slots_to_slots_nodes(self, mapping):
        """
        Converts a mapping of
        {id: <node>, slots: (slot1, slot2)}
        to
        {slot1: <node>, slot2: <node>}

        Operation is expensive so use with caution
        """
        out = {}
        for node in mapping:
            for slot in node['slots']:
                out[str(slot)] = node['id']
        return out

    async def cluster_addslots(self, node_id, *slots):
        """
        Assign new hash slots to receiving node

        Sends to specefied node
        """
        return await self.execute_command('CLUSTER ADDSLOTS', *slots, node_id=node_id)

    async def cluster_count_failure_report(self, node_id=''):
        """
        Return the number of failure reports active for a given node

        Sends to specefied node
        """
        return await self.execute_command('CLUSTER COUNT-FAILURE-REPORTS', node_id=node_id)

    async def cluster_countkeysinslot(self, slot_id):
        """
        Return the number of local keys in the specified hash slot

        Send to node based on specefied slot_id
        """
        return await self.execute_command('CLUSTER COUNTKEYSINSLOT', slot_id)

    async def cluster_delslots(self, *slots):
        """
        Set hash slots as unbound in the cluster.
        It determines by it self what node the slot is in and sends it there

        Returns a list of the results for each processed slot.
        """
        cluster_nodes = self._nodes_slots_to_slots_nodes(await self.cluster_nodes())
        res = list()
        for slot in slots:
            res.append(await self.execute_command('CLUSTER DELSLOTS', slot, node_id=cluster_nodes[slot]))
        return res

    async def cluster_failover(self, node_id, option):
        """
        Forces a slave to perform a manual failover of its master

        Sends to specefied node
        """
        if not isinstance(option, str) or option.upper() not in {'FORCE', 'TAKEOVER'}:
            raise ClusterError('Wrong option provided')
        return await self.execute_command('CLUSTER FAILOVER', option, node_id=node_id)

    async def cluster_forget(self, node_id):
        """
        remove a node via its node ID from the set of known nodes
        of the Redis Cluster node receiving the command

        Sends to all nodes in the cluster
        """
        return await self.execute_command('CLUSTER FORGET', node_id)

    async def cluster_info(self):
        """
        Provides info about Redis Cluster node state

        Sends to random node in the cluster
        """
        return await self.execute_command('CLUSTER INFO')

    async def cluster_keyslot(self, name):
        """
        Returns the hash slot of the specified key

        Sends to random node in the cluster
        """
        return await self.execute_command('CLUSTER KEYSLOT', name)

    async def cluster_meet(self, node_id, host, port):
        """
        Force a node cluster to handshake with another node.

        Sends to specefied node
        """
        return await self.execute_command('CLUSTER MEET', host, port, node_id=node_id)

    async def cluster_nodes(self):
        """
        Force a node cluster to handshake with another node

        Sends to random node in the cluster
        """
        return await self.execute_command('CLUSTER NODES')

    async def cluster_replicate(self, target_node_id):
        """
        Reconfigure a node as a slave of the specified master node

        Sends to specefied node
        """
        return await self.execute_command('CLUSTER REPLICATE', target_node_id)

    async def cluster_reset(self, node_id, soft=True):
        """
        Reset a Redis Cluster node

        If 'soft' is True then it will send 'SOFT' argument
        If 'soft' is False then it will send 'HARD' argument

        Sends to specefied node
        """
        option = 'SOFT' if soft else 'HARD'
        return await self.execute_command('CLUSTER RESET', option, node_id=node_id)

    async def cluster_reset_all_nodes(self, soft=True):
        """
        Send CLUSTER RESET to all nodes in the cluster

        If 'soft' is True then it will send 'SOFT' argument
        If 'soft' is False then it will send 'HARD' argument

        Sends to all nodes in the cluster
        """
        option = 'SOFT' if soft else 'HARD'
        res = list()
        for node in await self.cluster_nodes():
            res.append(
                await self.execute_command(
                    'CLUSTER RESET', option, node_id=node['id']
                ))
        return res

    async def cluster_save_config(self):
        """
        Forces the node to save cluster state on disk

        Sends to all nodes in the cluster
        """
        return await self.execute_command('CLUSTER SAVECONFIG')

    async def cluster_set_config_epoch(self, node_id, epoch):
        """
        Set the configuration epoch in a new node

        Sends to specefied node
        """
        return await self.execute_command('CLUSTER SET-CONFIG-EPOCH', epoch, node_id=node_id)


    async def cluster_setslot(self, node_id, slot_id, state):
        """
        Bind an hash slot to a specific node

        Sends to specified node
        """
        if state.upper() in {'IMPORTING', 'MIGRATING', 'NODE'} and node_id is not None:
            return await self.execute_command('CLUSTER SETSLOT', slot_id, state, node_id)
        elif state.upper() == 'STABLE':
            return await self.execute_command('CLUSTER SETSLOT', slot_id, 'STABLE')
        else:
            raise RedisError('Invalid slot state: {0}'.format(state))

    async def cluster_get_keys_in_slot(self, slot_id, count):
        """
        Return local key names in the specified hash slot
        Sends to specified node
        """
        return await self.execute_command('CLUSTER GETKEYSINSLOT', slot_id, count)

    async def cluster_slaves(self, target_node_id):
        """
        Force a node cluster to handshake with another node

        Sends to targeted cluster node
        """
        return await self.execute_command('CLUSTER SLAVES', target_node_id)

    async def cluster_slots(self):
        """
        Get array of Cluster slot to node mappings

        Sends to random node in the cluster
        """
        return await self.execute_command('CLUSTER SLOTS') class ClusterConnectionCommandMixin(ConnectionCommandMixin):

    NODES_FLAGS = {
        'PING': NodeFlag.ALL_NODES,
        'ECHO': NodeFlag.ALL_NODES
    }

    RESULT_CALLBACKS = {
        'ECHO': lambda res: res,
        'PING': lambda res: res
    } class ConnectionCommandMixin:

    RESPONSE_CALLBACKS = {
        'AUTH': bool,
        'PING': lambda r: nativestr(r) == 'PONG',
        'SELECT': bool_ok,
    }

    async def echo(self, value):
        "Echo the string back from the server"
        return await self.execute_command('ECHO', value)

    async def ping(self):
        "Ping the Redis server"
        return await self.execute_command('PING') class ExtraCommandMixin:

    RESPONSE_CALLBACKS = {}

    def cache(self, name, cache_class=Cache,
              identity_generator_class=IdentityGenerator,
              compressor_class=Compressor,
              serializer_class=Serializer, *args, **kwargs):
        """
        Return a cache object using default identity generator,
        serializer and compressor.

        ``name`` is used to identify the series of your cache
        ``cache_class`` Cache is for normal use and HerdCache
        is used in case of Thundering Herd Problem
        ``identity_generator_class`` is the class used to generate
        the real unique key in cache, can be overwritten to
        meet your special needs. It should provide `generate` API
        ``compressor_class`` is the class used to compress cache in redis,
        can be overwritten with API `compress` and `decompress` retained.
        ``serializer_class`` is the class used to serialize
        content before compress, can be overwritten with API
        `serialize` and `deserialize` retained.
        """
        return cache_class(self, app=name,
                           identity_generator_class=identity_generator_class,
                           compressor_class=compressor_class,
                           serializer_class=serializer_class,
                           *args, **kwargs)

    def lock(self, name, timeout=None, sleep=0.1, blocking_timeout=None,
             lock_class=None, thread_local=True):
        """
        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_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.

        ``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.        """
        if lock_class is None:
            if self._use_lua_lock is None:
                # the first time .lock() is called, determine if we can use
                # Lua by attempting to register the necessary scripts
                try:
                    LuaLock.register_scripts(self)
                    self._use_lua_lock = True
                except ResponseError:
                    self._use_lua_lock = False
            lock_class = self._use_lua_lock and LuaLock or Lock
        return lock_class(self, name, timeout=timeout, sleep=sleep,
                          blocking_timeout=blocking_timeout,
                          thread_local=thread_local) class GeoCommandMixin:

    RESPONSE_CALLBACKS = {
        'GEOPOS': lambda r: list(map(lambda ll: (float(ll[0]),
                                                 float(ll[1]))
                                                 if ll is not None else None, r)),
        'GEOHASH': lambda r: list(r),
        'GEORADIUS': parse_georadius_generic,
        'GEORADIUSBYMEMBER': parse_georadius_generic,
        'GEODIST': float,
        'GEOADD': int
    }

    # GEO COMMANDS
    async def geoadd(self, name, *values):
        """
        Add the specified geospatial items to the specified key identified
        by the ``name`` argument. The Geospatial items are given as ordered
        members of the ``values`` argument, each item or place is formed by
        the triad latitude, longitude and name.
        """
        if len(values) % 3 != 0:
            raise RedisError("GEOADD requires places with lon, lat and name"
                             " values")
        return await self.execute_command('GEOADD', name, *values)

    async def geodist(self, name, place1, place2, unit=None):
        """
        Return the distance between ``place1`` and ``place2`` members of the
        ``name`` key.
        The units must be one of the following : m, km mi, ft. By async default
        meters are used.
        """
        pieces = [name, place1, place2]
        if unit and unit not in ('m', 'km', 'mi', 'ft'):
            raise RedisError("GEODIST invalid unit")
        elif unit:
            pieces.append(unit)
        return await self.execute_command('GEODIST', *pieces)

    async def geohash(self, name, *values):
        """
        Return the geo hash string for each item of ``values`` members of
        the specified key identified by the ``name``argument.
        """
        return await self.execute_command('GEOHASH', name, *values)

    async def geopos(self, name, *values):
        """
        Return the positions of each item of ``values`` as members of
        the specified key identified by the ``name``argument. Each position
        is represented by the pairs lon and lat.
        """
        return await self.execute_command('GEOPOS', name, *values)

    async def georadius(self, name, longitude, latitude, radius, unit=None,
                        withdist=False, withcoord=False, withhash=False, count=None,
                        sort=None, store=None, store_dist=None):
        """
        Return the members of the specified key identified by the
        ``name`` argument which are within the borders of the area specified
        with the ``latitude`` and ``longitude`` location and the maximum
        distance from the center specified by the ``radius`` value.

        The units must be one of the following : m, km mi, ft. By default

        ``withdist`` indicates to return the distances of each place.

        ``withcoord`` indicates to return the latitude and longitude of
        each place.

        ``withhash`` indicates to return the geohash string of each place.

        ``count`` indicates to return the number of elements up to N.

        ``sort`` indicates to return the places in a sorted way, ASC for
        nearest to fairest and DESC for fairest to nearest.

        ``store`` indicates to save the places names in a sorted set named
        with a specific key, each element of the destination sorted set is
        populated with the score got from the original geo sorted set.

        ``store_dist`` indicates to save the places names in a sorted set
        named with a specific key, instead of ``store`` the sorted set
        destination score is set with the distance.
        """
        return await self._georadiusgeneric('GEORADIUS',
                                            name, longitude, latitude, radius,
                                            unit=unit, withdist=withdist,
                                            withcoord=withcoord, withhash=withhash,
                                            count=count, sort=sort, store=store,
                                            store_dist=store_dist)

    async def georadiusbymember(self, name, member, radius, unit=None,
                                withdist=False, withcoord=False, withhash=False,
                                count=None, sort=None, store=None, store_dist=None):
        """
        This command is exactly like ``georadius`` with the sole difference
        that instead of taking, as the center of the area to query, a longitude
        and latitude value, it takes the name of a member already existing
        inside the geospatial index represented by the sorted set.
        """
        return await self._georadiusgeneric('GEORADIUSBYMEMBER',
                                            name, member, radius, unit=unit,
                                            withdist=withdist, withcoord=withcoord,
                                            withhash=withhash, count=count,
                                            sort=sort, store=store,
                                            store_dist=store_dist)

    async def _georadiusgeneric(self, command, *args, **kwargs):
        pieces = list(args)
        if kwargs['unit'] and kwargs['unit'] not in ('m', 'km', 'mi', 'ft'):
            raise RedisError("GEORADIUS invalid unit")
        elif kwargs['unit']:
            pieces.append(kwargs['unit'])
        else:
            pieces.append('m', )

        for token in ('withdist', 'withcoord', 'withhash'):
            if kwargs[token]:
                pieces.append(b(token.upper()))

        if kwargs['count']:
            pieces.extend([b('COUNT'), kwargs['count']])

        if kwargs['sort'] and kwargs['sort'] not in ('ASC', 'DESC'):
            raise RedisError("GEORADIUS invalid sort")
        elif kwargs['sort']:
            pieces.append(b(kwargs['sort']))

        if kwargs['store'] and kwargs['store_dist']:
            raise RedisError("GEORADIUS store and store_dist cant be set"
                             " together")

        if kwargs['store']:
            pieces.extend([b('STORE'), kwargs['store']])

        if kwargs['store_dist']:
            pieces.extend([b('STOREDIST'), kwargs['store_dist']])

        return await self.execute_command(command, *pieces, **kwargs) class ClusterHashCommandMixin(HashCommandMixin):

    RESULT_CALLBACKS = {
        'HSCAN': first_key
    } class HashCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict('HDEL HLEN', int),
        string_keys_to_dict('HEXISTS HMSET', bool),
        {
            'HGETALL': lambda r: r and pairs_to_dict(r) or {},
            'HINCRBYFLOAT': float,
            'HSCAN': parse_hscan,
        }
    )

    async def hdel(self, name, *keys):
        """Deletes ``keys`` from hash ``name``"""
        return await self.execute_command('HDEL', name, *keys)

    async def hexists(self, name, key):
        """
        Returns a boolean indicating if ``key`` exists within hash ``name``
        """
        return await self.execute_command('HEXISTS', name, key)

    async def hget(self, name, key):
        """Returns the value of ``key`` within the hash ``name``"""
        return await self.execute_command('HGET', name, key)

    async def hgetall(self, name):
        """Returns a Python dict of the hash's name/value pairs"""
        return await self.execute_command('HGETALL', name)

    async def hincrby(self, name, key, amount=1):
        """Increments the value of ``key`` in hash ``name`` by ``amount``"""
        return await self.execute_command('HINCRBY', name, key, amount)

    async def hincrbyfloat(self, name, key, amount=1.0):
        """
        Increments the value of ``key`` in hash ``name`` by floating
        ``amount``
        """
        return await self.execute_command('HINCRBYFLOAT', name, key, amount)

    async def hkeys(self, name):
        """Returns the list of keys within hash ``name``"""
        return await self.execute_command('HKEYS', name)

    async def hlen(self, name):
        """Returns the number of elements in hash ``name``"""
        return await self.execute_command('HLEN', name)

    async def hset(self, name, key, value):
        """
        Sets ``key`` to ``value`` within hash ``name``
        Returns 1 if HSET created a new field, otherwise 0
        """
        return await self.execute_command('HSET', name, key, value)

    async def hsetnx(self, name, key, value):
        """
        Sets ``key`` to ``value`` within hash ``name`` if ``key`` does not
        exist.  Returns 1 if HSETNX created a field, otherwise 0.
        """
        return await self.execute_command('HSETNX', name, key, value)

    async def hmset(self, name, mapping):
        """
        Sets key to value within hash ``name`` for each corresponding
        key and value from the ``mapping`` dict.
        """
        if not mapping:
            raise DataError("'hmset' with 'mapping' of length 0")
        items = []
        for pair in iteritems(mapping):
            items.extend(pair)
        return await self.execute_command('HMSET', name, *items)

    async def hmget(self, name, keys, *args):
        """Returns a list of values ordered identically to ``keys``"""
        args = list_or_args(keys, args)
        return await self.execute_command('HMGET', name, *args)

    async def hvals(self, name):
        """Returns the list of values within hash ``name``"""
        return await self.execute_command('HVALS', name)

    async def hscan(self, name, cursor=0, match=None, count=None):
        """
        Incrementallys return key/value slices in a hash. Also returns a
        cursor pointing to the scan position.

        ``match`` allows for filtering the keys by pattern

        ``count`` allows for hint the minimum number of returns
        """
        pieces = [name, cursor]
        if match is not None:
            pieces.extend([b('MATCH'), match])
        if count is not None:
            pieces.extend([b('COUNT'), count])
        return await self.execute_command('HSCAN', *pieces)

    async def hstrlen(self, name, key):
        """
        Returns the string length of the value associated
        with field in the hash stored at key.
        If the key or the field do not exist, 0 is returned.
        """
        return await self.execute_command('HSTRLEN', name, key) class ClusterHyperLogCommandMixin(HyperLogCommandMixin):

    async def pfmerge(self, dest, *sources):
        """
        Merge N different HyperLogLogs into a single one.

        Cluster impl:
            Very special implementation is required to make pfmerge() work
            But it works :]
            It works by first fetching all HLL objects that should be merged and
            move them to one hashslot so that pfmerge operation can be performed without
            any 'CROSSSLOT' error.
            After the PFMERGE operation is done then it will be moved to the correct location
            within the cluster and cleanup is done.

            This operation is no longer atomic because of all the operations that has to be done.
        """
        all_k = []

        # Fetch all HLL objects via GET and store them client side as strings
        all_hll_objects = list()
        for hll_key in sources:
            all_hll_objects.append(await self.get(hll_key))

        # Randomize a keyslot hash that should be used inside {} when doing SET
        random_hash_slot = self._random_id()

        # Special handling of dest variable if it allready exists, then it shold be included in the HLL merge
        # dest can exists anywhere in the cluster.
        dest_data = await self.get(dest)

        if dest_data:
            all_hll_objects.append(dest_data)

        # SET all stored HLL objects with SET {RandomHash}RandomKey hll_obj
        for hll_object in all_hll_objects:
            k = self._random_good_hashslot_key(random_hash_slot)
            all_k.append(k)
            await self.set(k, hll_object)

        # Do regular PFMERGE operation and store value in random key in {RandomHash}
        tmp_dest = self._random_good_hashslot_key(random_hash_slot)
        await self.execute_command("PFMERGE", tmp_dest, *all_k)

        # Do GET and SET so that result will be stored in the destination object any where in the cluster
        parsed_dest = await self.get(tmp_dest)
        await self.set(dest, parsed_dest)

        # Cleanup tmp variables
        await self.delete(tmp_dest)

        for k in all_k:
            await self.delete(k)

        return True

    def _random_good_hashslot_key(self, hashslot):
        """
        Generate a good random key with a low probability of collision between any other key.
        """
        random_id = "{{0}}{1}".format(hashslot, self._random_id())
        return random_id

    def _random_id(self, size=16, chars=string.ascii_uppercase + string.digits):
        """
        Generates a random id based on `size` and `chars` variable.

        By default it will generate a 16 character long string based on
        ascii uppercase letters and digits.
        """
        return ''.join(random.choice(chars) for _ in range(size)) class HyperLogCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict('PFADD PFCOUNT', int),
        {
            'PFMERGE': bool_ok,
        }
    )

    async def pfadd(self, name, *values):
        "Adds the specified elements to the specified HyperLogLog."
        return await self.execute_command('PFADD', name, *values)

    async def pfcount(self, *sources):
        """
        Return the approximated cardinality of
        the set observed by the HyperLogLog at key(s).
        """
        return await self.execute_command('PFCOUNT', *sources)

    async def pfmerge(self, dest, *sources):
        "Merge N different HyperLogLogs into a single one."
        return await self.execute_command('PFMERGE', dest, *sources) class ClusterKeysCommandMixin(KeysCommandMixin):

    NODES_FLAGS = dict_merge(
        {
        'MOVE': NodeFlag.BLOCKED,
        'RANDOMKEY': NodeFlag.RANDOM,
        'SCAN': NodeFlag.ALL_MASTERS,
        },
        list_keys_to_dict(
            ['KEYS'],
            NodeFlag.ALL_NODES
        )
    )

    RESULT_CALLBACKS = {
        'KEYS': merge_result,
        'RANDOMKEY': first_key,
        'SCAN': lambda res: res
    }

    async def rename(self, src, dst):
        """
        Rename key ``src`` to ``dst``

        Cluster impl:
            This operation is no longer atomic because each key must be querried
            then set in separate calls because they maybe will change cluster node
        """
        if src == dst:
            raise ResponseError("source and destination objects are the same")

        data = await self.dump(src)

        if data is None:
            raise ResponseError("no such key")

        ttl = await self.pttl(src)

        if ttl is None or ttl < 1:
            ttl = 0

        await self.delete(dst)
        await self.restore(dst, ttl, data)
        await self.delete(src)

        return True

    async def delete(self, *names):
        """
        "Delete one or more keys specified by ``names``"

        Cluster impl:
            Iterate all keys and send DELETE for each key.
            This will go a lot slower than a normal delete call in StrictRedis.

            Operation is no longer atomic.
        """
        count = 0

        for arg in names:
            count += await self.execute_command('DEL', arg)

        return count

    async def renamenx(self, src, dst):
        """
        Rename key ``src`` to ``dst`` if ``dst`` doesn't already exist

        Cluster impl:
            Check if dst key do not exists, then calls rename().

            Operation is no longer atomic.
        """
        if not await self.exists(dst):
            return await self.rename(src, dst)

        return False class KeysCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict(
            'EXISTS EXPIRE EXPIREAT '
            'MOVE PERSIST RENAMENX', bool
        ),
        {
            'DEL': int,
            'SORT': sort_return_tuples,
            'OBJECT': parse_object,
            'RANDOMKEY': lambda r: r and r or None,
            'SCAN': parse_scan,
            'RENAME': bool_ok,
        }
    )

    async def delete(self, *names):
        """Delete one or more keys specified by ``names``"""
        return await self.execute_command('DEL', *names)

    async def dump(self, name):
        """
        Return a serialized version of the value stored at the specified key.
        If key does not exist a nil bulk reply is returned.
        """
        return await self.execute_command('DUMP', name)

    async def exists(self, name):
        """Returns a boolean indicating whether key ``name`` exists"""
        return await self.execute_command('EXISTS', name)

    async def expire(self, name, time):
        """
        Set an expire flag on key ``name`` for ``time`` seconds. ``time``
        can be represented by an integer or a Python timedelta object.
        """
        if isinstance(time, datetime.timedelta):
            time = time.seconds + time.days * 24 * 3600
        return await self.execute_command('EXPIRE', name, time)

    async def expireat(self, name, when):
        """
        Set an expire flag on key ``name``. ``when`` can be represented
        as an integer indicating unix time or a Python datetime object.
        """
        if isinstance(when, datetime.datetime):
            when = int(mod_time.mktime(when.timetuple()))
        return await self.execute_command('EXPIREAT', name, when)

    async def keys(self, pattern='*'):
        """Returns a list of keys matching ``pattern``"""
        return await self.execute_command('KEYS', pattern)

    async def move(self, name, db):
        """Moves the key ``name`` to a different Redis database ``db``"""
        return await self.execute_command('MOVE', name, db)

    async def object(self, infotype, key):
        """Returns the encoding, idletime, or refcount about the key"""
        return await self.execute_command('OBJECT', infotype, key, infotype=infotype)

    async def persist(self, name):
        """Removes an expiration on ``name``"""
        return await self.execute_command('PERSIST', name)

    async def pexpire(self, name, time):
        """
        Set an expire flag on key ``name`` for ``time`` milliseconds.
        ``time`` can be represented by an integer or a Python timedelta
        object.
        """
        if isinstance(time, datetime.timedelta):
            ms = int(time.microseconds / 1000)
            time = (time.seconds + time.days * 24 * 3600) * 1000 + ms
        return await self.execute_command('PEXPIRE', name, time)

    async def pexpireat(self, name, when):
        """
        Set an expire flag on key ``name``. ``when`` can be represented
        as an integer representing unix time in milliseconds (unix time * 1000)
        or a Python datetime object.
        """
        if isinstance(when, datetime.datetime):
            ms = int(when.microsecond / 1000)
            when = int(mod_time.mktime(when.timetuple())) * 1000 + ms
        return await self.execute_command('PEXPIREAT', name, when)

    async def pttl(self, name):
        """
        Returns the number of milliseconds until the key ``name`` will expire
        """
        return await self.execute_command('PTTL', name)

    async def randomkey(self):
        """Returns the name of a random key"""
        return await self.execute_command('RANDOMKEY')

    async def rename(self, src, dst):
        """
        Renames key ``src`` to ``dst``
        """
        return await self.execute_command('RENAME', src, dst)

    async def renamenx(self, src, dst):
        """Renames key ``src`` to ``dst`` if ``dst`` doesn't already exist"""
        return await self.execute_command('RENAMENX', src, dst)

    async def restore(self, name, ttl, value, replace=False):
        """
        Creates a key using the provided serialized value, previously obtained
        using DUMP.
        """
        params = [name, ttl, value]
        if replace:
            params.append('REPLACE')
        return await self.execute_command('RESTORE', *params)

    async def sort(self, name, start=None, num=None, by=None, get=None,
             desc=False, alpha=False, store=None, groups=False):
        """
        Sorts and returns a list, set or sorted set at ``name``.

        ``start`` and ``num`` are for paginating sorted data

        ``by`` allows using an external key to weight and sort the items.
            Use an "*" to indicate where in the key the item value is located

        ``get`` is for returning items from external keys rather than the
            sorted data itself.  Use an "*" to indicate where int he key
            the item value is located

        ``desc`` is for reversing the sort

        ``alpha`` is for sorting lexicographically rather than numerically

        ``store`` is for storing the result of the sort into
            the key ``store``

        ``groups`` if set to True and if ``get`` contains at least two
            elements, sort will return a list of tuples, each containing the
            values fetched from the arguments to ``get``.

        """
        if (start is not None and num is None) or \
                (num is not None and start is None):
            raise RedisError("``start`` and ``num`` must both be specified")

        pieces = [name]
        if by is not None:
            pieces.append(b('BY'))
            pieces.append(by)
        if start is not None and num is not None:
            pieces.append(b('LIMIT'))
            pieces.append(start)
            pieces.append(num)
        if get is not None:
            # If get is a string assume we want to get a single value.
            # Otherwise assume it's an interable and we want to get multiple
            # values. We can't just iterate blindly because strings are
            # iterable.
            if isinstance(get, str):
                pieces.append(b('GET'))
                pieces.append(get)
            else:
                for g in get:
                    pieces.append(b('GET'))
                    pieces.append(g)
        if desc:
            pieces.append(b('DESC'))
        if alpha:
            pieces.append(b('ALPHA'))
        if store is not None:
            pieces.append(b('STORE'))
            pieces.append(store)

        if groups:
            if not get or isinstance(get, str) or len(get) < 2:
                raise DataError('when using "groups" the "get" argument '
                                'must be specified and contain at least '
                                'two keys')

        options = {'groups': len(get) if groups else None}
        return await self.execute_command('SORT', *pieces, **options)

    async def touch(self, keys):
        """
        Alters the last access time of a key(s).
        A key is ignored if it does not exist.
        """
        return await self.execute_command('TOUCH', *keys)

    async def ttl(self, name):
        """Returns the number of seconds until the key ``name`` will expire"""
        return await self.execute_command('TTL', name)

    async def type(self, name):
        """Returns the type of key ``name``"""
        return await self.execute_command('TYPE', name)

    async def unlink(self, *keys):
        """Removes the specified keys in a different thread, not blocking"""
        return await self.execute_command('UNLINK', *keys)

    async def wait(self, num_replicas, timeout):
        """
        Redis synchronous replication
        That returns the number of replicas that processed the query when
        we finally have at least ``num_replicas``, or when the ``timeout`` was
        reached.
        """
        return await self.execute_command('WAIT', num_replicas, timeout)

    async def scan(self, cursor=0, match=None, count=None):
        """
        Incrementally return lists of key names. Also return a cursor
        indicating the scan position.

        ``match`` allows for filtering the keys by pattern

        ``count`` allows for hint the minimum number of returns
        """
        pieces = [cursor]
        if match is not None:
            pieces.extend([b('MATCH'), match])
        if count is not None:
            pieces.extend([b('COUNT'), count])
        return await self.execute_command('SCAN', *pieces) class ClusterListsCommandMixin(ListsCommandMixin):

    async def brpoplpush(self, src, dst, timeout=0):
        """
        Pops a value off the tail of ``src``, push it on the head of ``dst``
        and then return it.

        This command blocks until a value is in ``src`` or until ``timeout``
        seconds elapse, whichever is first. A ``timeout`` value of 0 blocks
        forever.

        Cluster impl:
            Call brpop() then send the result into lpush()

            Operation is no longer atomic.
        """
        try:
            value = await self.brpop(src, timeout=timeout)
            if value is None:
                return None
        except TimeoutError:
            # Timeout was reached
            return None

        await self.lpush(dst, value[1])
        return value[1]

    async def rpoplpush(self, src, dst):
        """
        RPOP a value off of the ``src`` list and atomically LPUSH it
        on to the ``dst`` list.  Returns the value.

        Cluster impl:
            Call rpop() then send the result into lpush()

            Operation is no longer atomic.
        """
        value = await self.rpop(src)

        if value:
            await self.lpush(dst, value)
            return value

        return None

    async def sort(self, name, start=None, num=None, by=None, get=None, desc=False, alpha=False, store=None, groups=None):
        """Sorts and returns a list, set or sorted set at ``name``.

        :start: and :num:
            are for paging through the sorted data

        :by:
            allows using an external key to weight and sort the items.
            Use an "*" to indicate where in the key the item value is located

        :get:
            is for returning items from external keys rather than the
            sorted data itself.  Use an "*" to indicate where int he key
            the item value is located

        :desc:
            is for reversing the sort

        :alpha:
            is for sorting lexicographically rather than numerically

        :store:
            is for storing the result of the sort into the key `store`

        ClusterImpl:
            A full implementation of the server side sort mechanics because many of the
            options work on multiple keys that can exist on multiple servers.
        """
        if (start is None and num is not None) or \
                (start is not None and num is None):
            raise RedisError("RedisError: ``start`` and ``num`` must both be specified")
        try:
            data_type = b(await self.type(name))

            if data_type == b("none"):
                return []
            elif data_type == b("set"):
                data = list(await self.smembers(name))[:]
            elif data_type == b("list"):
                data = await self.lrange(name, 0, -1)
            else:
                raise RedisClusterException("Unable to sort data type : {0}".format(data_type))
            if by is not None:
                # _sort_using_by_arg mutates data so we don't
                # need need a return value.
                data = await self._sort_using_by_arg(data, by, alpha)
            elif not alpha:
                data.sort(key=self._strtod_key_func)
            else:
                data.sort()
            if desc:
                data = data[::-1]
            if not (start is None and num is None):
                data = data[start:start + num]

            if get:
                data = await self._retrive_data_from_sort(data, get)

            if store is not None:
                if data_type == b("set"):
                    await self.delete(store)
                    await self.rpush(store, *data)
                elif data_type == b("list"):
                    await self.delete(store)
                    await self.rpush(store, *data)
                else:
                    raise RedisClusterException("Unable to store sorted data for data type : {0}".format(data_type))

                return len(data)

            if groups:
                if not get or isinstance(get, str) or len(get) < 2:
                    raise DataError('when using "groups" the "get" argument '
                                    'must be specified and contain at least '
                                    'two keys')
                n = len(get)
                return list(zip(*[data[i::n] for i in range(n)]))
            else:
                return data
        except KeyError:
            return []

    async def _retrive_data_from_sort(self, data, get):
        """
        Used by sort()
        """
        if get is not None:
            if isinstance(get, str):
                get = [get]
            new_data = []
            for k in data:
                for g in get:
                    single_item = await self._get_single_item(k, g)
                    new_data.append(single_item)
            data = new_data
        return data

    async def _get_single_item(self, k, g):
        """
        Used by sort()
        """
        if getattr(k, "decode", None):
            k = k.decode("utf-8")

        if '*' in g:
            g = g.replace('*', k)
            if '->' in g:
                key, hash_key = g.split('->')
                single_item = await self.get(key, {}).get(hash_key)
            else:
                single_item = await self.get(g)
        elif '#' in g:
            single_item = k
        else:
            single_item = None
        return b(single_item)

    def _strtod_key_func(self, arg):
        """
        Used by sort()
        """
        return float(arg)

    async def _sort_using_by_arg(self, data, by, alpha):
        """
        Used by sort()
        """
        if getattr(by, "decode", None):
            by = by.decode("utf-8")

        async def _by_key(arg):
            if getattr(arg, "decode", None):
                arg = arg.decode("utf-8")

            key = by.replace('*', arg)
            if '->' in by:
                key, hash_key = key.split('->')
                v = await self.hget(key, hash_key)
                if alpha:
                    return v
                else:
                    return float(v)
            else:
                return await self.get(key)
        sorted_data = []
        for d in data:
            sorted_data.append((d, await _by_key(d)))
        return [x[0] for x in sorted(sorted_data, key=lambda x: x[1])] class ListsCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict(
            'BLPOP BRPOP',
            lambda r: r and tuple(r) or None
        ),
        string_keys_to_dict(
            # these return OK, or int if redis-server is >=1.3.4
            'LPUSH RPUSH',
            lambda r: isinstance(r, int) and r or nativestr(r) == 'OK'
        ),
        string_keys_to_dict('LSET LTRIM', bool_ok),
        string_keys_to_dict('LINSERT LLEN LPUSHX RPUSHX', int),
    )

    async def blpop(self, keys, timeout=0):
        """
        LPOP a value off of the first non-empty list
        named in the ``keys`` list.

        If none of the lists in ``keys`` has a value to LPOP, then block
        for ``timeout`` seconds, or until a value gets pushed on to one
        of the lists.

        If timeout is 0, then block indefinitely.
        """
        if timeout is None:
            timeout = 0
        if isinstance(keys, str):
            keys = [keys]
        else:
            keys = list(keys)
        keys.append(timeout)
        return await self.execute_command('BLPOP', *keys)

    async def brpop(self, keys, timeout=0):
        """
        RPOP a value off of the first non-empty list
        named in the ``keys`` list.

        If none of the lists in ``keys`` has a value to LPOP, then block
        for ``timeout`` seconds, or until a value gets pushed on to one
        of the lists.

        If timeout is 0, then block indefinitely.
        """
        if timeout is None:
            timeout = 0
        if isinstance(keys, str):
            keys = [keys]
        else:
            keys = list(keys)
        keys.append(timeout)
        return await self.execute_command('BRPOP', *keys)

    async def brpoplpush(self, src, dst, timeout=0):
        """
        Pop a value off the tail of ``src``, push it on the head of ``dst``
        and then return it.

        This command blocks until a value is in ``src`` or until ``timeout``
        seconds elapse, whichever is first. A ``timeout`` value of 0 blocks
        forever.
        """
        if timeout is None:
            timeout = 0
        return await self.execute_command('BRPOPLPUSH', src, dst, timeout)

    async def lindex(self, name, index):
        """
        Return the item from list ``name`` at position ``index``

        Negative indexes are supported and will return an item at the
        end of the list
        """
        return await self.execute_command('LINDEX', name, index)

    async def linsert(self, name, where, refvalue, value):
        """
        Insert ``value`` in list ``name`` either immediately before or after
        [``where``] ``refvalue``

        Returns the new length of the list on success or -1 if ``refvalue``
        is not in the list.
        """
        return await self.execute_command('LINSERT', name, where, refvalue, value)

    async def llen(self, name):
        """Returns the length of the list ``name``"""
        return await self.execute_command('LLEN', name)

    async def lpop(self, name):
        """RemoveS and returns the first item of the list ``name``"""
        return await self.execute_command('LPOP', name)

    async def lpush(self, name, *values):
        """Pushes ``values`` onto the head of the list ``name``"""
        return await self.execute_command('LPUSH', name, *values)

    async def lpushx(self, name, value):
        """
        Pushes ``value`` onto the head of the list ``name`` if ``name`` exists
        """
        return await self.execute_command('LPUSHX', name, value)

    async def lrange(self, name, start, end):
        """
        Returns a slice of the list ``name`` between
        position ``start`` and ``end``

        ``start`` and ``end`` can be negative numbers just like
        Python slicing notation
        """
        return await self.execute_command('LRANGE', name, start, end)

    async def lrem(self, name, count, value):
        """
        Removes the first ``count`` occurrences of elements equal to ``value``
        from the list stored at ``name``.

        The count argument influences the operation in the following ways:
            count > 0: Remove elements equal to value moving from head to tail.
            count < 0: Remove elements equal to value moving from tail to head.
            count = 0: Remove all elements equal to value.
        """
        return await self.execute_command('LREM', name, count, value)

    async def lset(self, name, index, value):
        """Sets ``position`` of list ``name`` to ``value``"""
        return await self.execute_command('LSET', name, index, value)

    async def ltrim(self, name, start, end):
        """
        Trims the list ``name``, removing all values not within the slice
        between ``start`` and ``end``

        ``start`` and ``end`` can be negative numbers just like
        Python slicing notation
        """
        return await self.execute_command('LTRIM', name, start, end)

    async def rpop(self, name):
        """Removes and return the last item of the list ``name``"""
        return await self.execute_command('RPOP', name)

    async def rpoplpush(self, src, dst):
        """
        RPOP a value off of the ``src`` list and atomically LPUSH it
        on to the ``dst`` list.  Returns the value.
        """
        return await self.execute_command('RPOPLPUSH', src, dst)

    async def rpush(self, name, *values):
        """Pushes ``values`` onto the tail of the list ``name``"""
        return await self.execute_command('RPUSH', name, *values)

    async def rpushx(self, name, value):
        """
        Pushes ``value`` onto the tail of the list ``name`` if ``name`` exists
        """
        return await self.execute_command('RPUSHX', name, value) class CLusterPubSubCommandMixin(PubSubCommandMixin):
    
    NODES_FLAGS = dict_merge(
        list_keys_to_dict(
            ['PUBSUB CHANNELS', 'PUBSUB NUMSUB', 'PUBSUB NUMPAT'],
            NodeFlag.ALL_NODES
        )
    )

    RESULT_CALLBACKS = dict_merge(
        list_keys_to_dict([
            "PUBSUB CHANNELS",
        ], parse_cluster_pubsub_channels),
        list_keys_to_dict([
            "PUBSUB NUMSUB",
        ], parse_cluster_pubsub_numsub),
        list_keys_to_dict([
            "PUBSUB NUMPAT",
        ], parse_cluster_pubsub_numpat),
    )

    def pubsub(self, **kwargs):
        return ClusterPubSub(self.connection_pool, **kwargs) class PubSubCommandMixin:

    RESPONSE_CALLBACKS = {
        'PUBSUB NUMSUB': parse_pubsub_numsub,
    }

    def pubsub(self, **kwargs):
        """
        Return a Publish/Subscribe object. With this object, you can
        subscribe to channels and listen for messages that get published to
        them.
        """
        return PubSub(self.connection_pool, **kwargs)

    async def publish(self, channel, message):
        """
        Publish ``message`` on ``channel``.
        Returns the number of subscribers the message was delivered to.
        """
        return await self.execute_command('PUBLISH', channel, message)

    async def pubsub_channels(self, pattern='*'):
        """
        Return a list of channels that have at least one subscriber
        """
        return await self.execute_command('PUBSUB CHANNELS', pattern)

    async def pubsub_numpat(self):
        """
        Returns the number of subscriptions to patterns
        """
        return await self.execute_command('PUBSUB NUMPAT')

    async def pubsub_numsub(self, *args):
        """
        Return a list of (channel, number of subscribers) tuples
        for each channel given in ``*args``
        """
        return await self.execute_command('PUBSUB NUMSUB', *args) class ClusterScriptingCommandMixin(ScriptingCommandMixin):

    NODES_FLAGS = dict_merge(
        {
        'SCRIPT KILL': NodeFlag.BLOCKED
        },
        list_keys_to_dict(
            ["SCRIPT LOAD", "SCRIPT FLUSH", "SCRIPT EXISTS",], NodeFlag.ALL_MASTERS
        )
    )

    RESULT_CALLBACKS = {
        "SCRIPT LOAD": lambda res: list(res.values()).pop(),
        "SCRIPT EXISTS": lambda res: [all(k) for k in zip(*res.values())],
        "SCRIPT FLUSH": lambda res: all(res.values())
    } class ScriptingCommandMixin:

    RESPONSE_CALLBACKS = {
        'SCRIPT EXISTS': lambda r: list(map(bool, r)),
        'SCRIPT FLUSH': bool_ok,
        'SCRIPT KILL': bool_ok,
        'SCRIPT LOAD': nativestr,
    }

    async def eval(self, script, numkeys, *keys_and_args):
        """
        Execute the Lua ``script``, specifying the ``numkeys`` the script
        will touch and the key names and argument values in ``keys_and_args``.
        Returns the result of the script.

        In practice, use the object returned by ``register_script``. This
        function exists purely for Redis API completion.
        """
        return await self.execute_command('EVAL', script, numkeys, *keys_and_args)

    async def evalsha(self, sha, numkeys, *keys_and_args):
        """
        Use the ``sha`` to execute a Lua script already registered via EVAL
        or SCRIPT LOAD. Specify the ``numkeys`` the script will touch and the
        key names and argument values in ``keys_and_args``. Returns the result
        of the script.

        In practice, use the object returned by ``register_script``. This
        function exists purely for Redis API completion.
        """
        return await self.execute_command('EVALSHA', sha, numkeys, *keys_and_args)

    async def script_exists(self, *args):
        """
        Check if a script exists in the script cache by specifying the SHAs of
        each script as ``args``. Returns a list of boolean values indicating if
        if each already script exists in the cache.
        """
        return await self.execute_command('SCRIPT EXISTS', *args)

    async def script_flush(self):
        """Flushes all scripts from the script cache"""
        return await self.execute_command('SCRIPT FLUSH')

    async def script_kill(self):
        """Kills the currently executing Lua script"""
        return await self.execute_command('SCRIPT KILL')

    async def script_load(self, script):
        """Loads a Lua ``script`` into the script cache. Returns the SHA."""
        return await self.execute_command('SCRIPT LOAD', script)

    def register_script(self, script):
        """
        Registers a Lua ``script`` specifying the ``keys`` it will touch.
        Returns a Script object that is callable and hides the complexity of
        dealing with scripts, keys, and shas. This is the preferred way of
        working with Lua scripts.
        """
        from aredis.scripting import Script
        return Script(self, script) class ClusterSentinelCommands(SentinelCommandMixin):
    NODES_FLAGS = dict_merge(
        list_keys_to_dict(
            ['SENTINEL GET-MASTER-ADDR-BY-NAME', 'SENTINEL MASTER', 'SENTINEL MASTERS',
             'SENTINEL MONITOR', 'SENTINEL REMOVE', 'SENTINEL SENTINELS', 'SENTINEL SET',
             'SENTINEL SLAVES'], NodeFlag.BLOCKED
        )
    ) class SentinelCommandMixin:
    RESPONSE_CALLBACKS = {
        'SENTINEL GET-MASTER-ADDR-BY-NAME': parse_sentinel_get_master,
        'SENTINEL MASTER': parse_sentinel_master,
        'SENTINEL MASTERS': parse_sentinel_masters,
        'SENTINEL MONITOR': bool_ok,
        'SENTINEL REMOVE': bool_ok,
        'SENTINEL SENTINELS': parse_sentinel_slaves_and_sentinels,
        'SENTINEL SET': bool_ok,
        'SENTINEL SLAVES': parse_sentinel_slaves_and_sentinels,
    }

    async def sentinel(self, *args):
        """Redis Sentinel's SENTINEL command."""
        warnings.warn(DeprecationWarning('Use the individual sentinel_* methods'))

    async def sentinel_get_master_addr_by_name(self, service_name):
        """Returns a (host, port) pair for the given ``service_name``"""
        return await self.execute_command('SENTINEL GET-MASTER-ADDR-BY-NAME',
                                          service_name)

    async def sentinel_master(self, service_name):
        """Returns a dictionary containing the specified masters state."""
        return await self.execute_command('SENTINEL MASTER', service_name)

    async def sentinel_masters(self):
        """Returns a list of dictionaries containing each master's state."""
        return await self.execute_command('SENTINEL MASTERS')

    async def sentinel_monitor(self, name, ip, port, quorum):
        """Adds a new master to Sentinel to be monitored"""
        return await self.execute_command('SENTINEL MONITOR', name, ip, port, quorum)

    async def sentinel_remove(self, name):
        """Removes a master from Sentinel's monitoring"""
        return await self.execute_command('SENTINEL REMOVE', name)

    async def sentinel_sentinels(self, service_name):
        """Returns a list of sentinels for ``service_name``"""
        return await self.execute_command('SENTINEL SENTINELS', service_name)

    async def sentinel_set(self, name, option, value):
        """Sets Sentinel monitoring parameters for a given master"""
        return await self.execute_command('SENTINEL SET', name, option, value)

    async def sentinel_slaves(self, service_name):
        """Returns a list of slaves for ``service_name``"""
        return await self.execute_command('SENTINEL SLAVES', service_name) class ClusterServerCommandMixin(ServerCommandMixin):
    NODES_FLAGS = dict_merge(
        list_keys_to_dict(
            ['SHUTDOWN', 'SLAVEOF', 'CLIENT SETNAME'],
            NodeFlag.BLOCKED
        ),
        list_keys_to_dict(
            ['FLUSHALL', 'FLUSHDB'],
            NodeFlag.ALL_MASTERS
        ),
        list_keys_to_dict(
            ['SLOWLOG LEN', 'SLOWLOG RESET', 'SLOWLOG GET',
             'TIME', 'SAVE', 'LASTSAVE', 'DBSIZE',
             'CONFIG RESETSTAT', 'CONFIG REWRITE',
             'CONFIG GET', 'CONFIG SET', 'CLIENT KILL',
             'CLIENT LIST', 'CLIENT GETNAME', 'INFO',
             'BGSAVE', 'BGREWRITEAOF'],
            NodeFlag.ALL_NODES
        )
    )

    RESULT_CALLBACKS = dict_merge(
        list_keys_to_dict(
            ['CONFIG GET', 'CONFIG SET', 'SLOWLOG GET',
             'CLIENT KILL', 'INFO', 'BGREWRITEAOF',
             'BGSAVE', 'CLIENT LIST', 'CLIENT GETNAME',
             'CONFIG RESETSTAT', 'CONFIG REWRITE', 'DBSIZE',
             'LASTSAVE', 'SAVE', 'SLOWLOG LEN',
             'SLOWLOG RESET', 'TIME', 'FLUSHALL',
             'FLUSHDB'],
            lambda res: res
        )
    ) class ServerCommandMixin:
    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict('BGREWRITEAOF BGSAVE', lambda r: True),
        string_keys_to_dict(
            'FLUSHALL FLUSHDB SAVE '
            'SHUTDOWN SLAVEOF', bool_ok
        ),
        {
            'ROLE': parse_role,
            'SLOWLOG GET': parse_slowlog_get,
            'SLOWLOG LEN': int,
            'SLOWLOG RESET': bool_ok,
            'CLIENT GETNAME': lambda r: r and nativestr(r),
            'CLIENT KILL': bool_ok,
            'CLIENT LIST': parse_client_list,
            'CLIENT SETNAME': bool_ok,
            'CLIENT PAUSE': bool_ok,
            'CONFIG GET': parse_config_get,
            'CONFIG RESETSTAT': bool_ok,
            'CONFIG SET': bool_ok,
            'DEBUG OBJECT': parse_debug_object,
            'INFO': parse_info,
            'LASTSAVE': timestamp_to_datetime,
            'TIME': lambda x: (int(x[0]), int(x[1])),
        }
    )

    async def bgrewriteaof(self):
        """Tell the Redis server to rewrite the AOF file from data in memory"""
        return await self.execute_command('BGREWRITEAOF')

    async def bgsave(self):
        """
        Tells the Redis server to save its data to disk.  Unlike save(),
        this method is asynchronous and returns immediately.
        """
        return await self.execute_command('BGSAVE')

    async def client_kill(self, address):
        """Disconnects the client at ``address`` (ip:port)"""
        return await self.execute_command('CLIENT KILL', address)

    async def client_list(self):
        """Returns a list of currently connected clients"""
        return await self.execute_command('CLIENT LIST')

    async def client_getname(self):
        """Returns the current connection name"""
        return await self.execute_command('CLIENT GETNAME')

    async def client_setname(self, name):
        """Sets the current connection name"""
        return await self.execute_command('CLIENT SETNAME', name)

    async def client_pause(self, timeout=0):
        """
        Suspends all the Redis clients for the specified amount of time
        (in milliseconds).
        """
        return await self.execute_command('CLIENT PAUSE', timeout)

    async def config_get(self, pattern="*"):
        """Returns a dictionary of configuration based on the ``pattern``"""
        return await self.execute_command('CONFIG GET', pattern)

    async def config_set(self, name, value):
        """Sets config item ``name`` to ``value``"""
        return await self.execute_command('CONFIG SET', name, value)

    async def config_resetstat(self):
        """Resets runtime statistics"""
        return await self.execute_command('CONFIG RESETSTAT')

    async def config_rewrite(self):
        """
        Rewrites config file with the minimal change to reflect running config
        """
        return await self.execute_command('CONFIG REWRITE')

    async def dbsize(self):
        """Returns the number of keys in the current database"""
        return await self.execute_command('DBSIZE')

    async def debug_object(self, key):
        """Returns version specific meta information about a given key"""
        return await self.execute_command('DEBUG OBJECT', key)

    async def flushall(self):
        """Deletes all keys in all databases on the current host"""
        return await self.execute_command('FLUSHALL')

    async def flushdb(self):
        """Deletes all keys in the current database"""
        return await self.execute_command('FLUSHDB')

    async def info(self, section=None):
        """
        Returns a dictionary containing information about the Redis server

        The ``section`` option can be used to select a specific section
        of information

        The section option is not supported by older versions of Redis Server,
        and will generate ResponseError
        """
        if section is None:
            return await self.execute_command('INFO')
        else:
            return await self.execute_command('INFO', section)

    async def lastsave(self):
        """
        Returns a Python datetime object representing the last time the
        Redis database was saved to disk
        """
        return await self.execute_command('LASTSAVE')

    async def save(self):
        """
        Tells the Redis server to save its data to disk,
        blocking until the save is complete
        """
        return await self.execute_command('SAVE')

    async def shutdown(self):
        """Stops Redis server"""
        try:
            await self.execute_command('SHUTDOWN')
        except ConnectionError:
            # a ConnectionError here is expected
            return
        raise RedisError("SHUTDOWN seems to have failed.")

    async def slaveof(self, host=None, port=None):
        """
        Sets the server to be a replicated slave of the instance identified
        by the ``host`` and ``port``. If called without arguments, the
        instance is promoted to a master instead.
        """
        if host is None and port is None:
            return await self.execute_command('SLAVEOF', b('NO'), b('ONE'))
        return await self.execute_command('SLAVEOF', host, port)

    async def slowlog_get(self, num=None):
        """
        Gets the entries from the slowlog. If ``num`` is specified, get the
        most recent ``num`` items.
        """
        args = ['SLOWLOG GET']
        if num is not None:
            args.append(num)
        return await self.execute_command(*args)

    async def slowlog_len(self):
        """Gets the number of items in the slowlog"""
        return await self.execute_command('SLOWLOG LEN')

    async def slowlog_reset(self):
        """Removes all items in the slowlog"""
        return await self.execute_command('SLOWLOG RESET')

    async def time(self):
        """
        Returns the server time as a 2-item tuple of ints:
        (seconds since epoch, microseconds into this second).
        """
        return await self.execute_command('TIME')

    async def role(self):
        """
        Provides information on the role of a Redis instance in the context of replication,
        by returning if the instance is currently a master, slave, or sentinel.
        The command also returns additional information about the state of the replication
        (if the role is master or slave)
        or the list of monitored master names (if the role is sentinel).
        :return:
        """
        return await self.execute_command('ROLE') class ClusterSetsCommandMixin(SetsCommandMixin):

    RESULT_CALLBACKS = {
        'SSCAN': first_key
    }


    ###
    # Set commands

    async def sdiff(self, keys, *args):
        """
        Returns the difference of sets specified by ``keys``

        Cluster impl:
            Querry all keys and diff all sets and return result
        """
        k = list_or_args(keys, args)
        res = await self.smembers(k[0])

        for arg in k[1:]:
            res -= await self.smembers(arg)

        return res

    async def sdiffstore(self, dest, keys, *args):
        """
        Stores the difference of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.
        Overwrites dest key if it exists.

        Cluster impl:
            Use sdiff() --> Delete dest key --> store result in dest key
        """
        res = await self.sdiff(keys, *args)
        await self.delete(dest)

        if not res:
            return 0
        return await self.sadd(dest, *res)

    async def sinter(self, keys, *args):
        """
        Returns the intersection of sets specified by ``keys``

        Cluster impl:
            Querry all keys, intersection and return result
        """
        k = list_or_args(keys, args)
        res = await self.smembers(k[0])

        for arg in k[1:]:
            res &= await self.smembers(arg)

        return res

    async def sinterstore(self, dest, keys, *args):
        """
        Stores the intersection of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.

        Cluster impl:
            Use sinter() --> Delete dest key --> store result in dest key
        """
        res = await self.sinter(keys, *args)
        await self.delete(dest)

        if res:
            await self.sadd(dest, *res)
            return len(res)
        else:
            return 0

    async def smove(self, src, dst, value):
        """
        Moves ``value`` from set ``src`` to set ``dst`` atomically

        Cluster impl:
            SMEMBERS --> SREM --> SADD. Function is no longer atomic.
        """
        res = await self.srem(src, value)

        # Only add the element if existed in src set
        if res == 1:
            await self.sadd(dst, value)

        return res

    async def sunion(self, keys, *args):
        """
        Returns the union of sets specified by ``keys``

        Cluster impl:
            Querry all keys, union and return result

            Operation is no longer atomic.
        """
        k = list_or_args(keys, args)
        res = await self.smembers(k[0])

        for arg in k[1:]:
            res |= await self.smembers(arg)

        return res

    async def sunionstore(self, dest, keys, *args):
        """
        Stores the union of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.

        Cluster impl:
            Use sunion() --> Dlete dest key --> store result in dest key

            Operation is no longer atomic.
        """
        res = await self.sunion(keys, *args)
        await self.delete(dest)

        return await self.sadd(dest, *res) class SetsCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict(
            'SADD SCARD SDIFFSTORE '
            'SETRANGE SINTERSTORE '
            'SREM SUNIONSTORE', int
        ),
        string_keys_to_dict(
            'SISMEMBER SMOVE', bool
        ),
        string_keys_to_dict(
            'SDIFF SINTER SMEMBERS SUNION',
            lambda r: r and set(r) or set()
        ),
        {
            'SSCAN': parse_sscan,
        }
    )

    async def sadd(self, name, *values):
        """Adds ``value(s)`` to set ``name``"""
        return await self.execute_command('SADD', name, *values)

    async def scard(self, name):
        """Returns the number of elements in set ``name``"""
        return await self.execute_command('SCARD', name)

    async def sdiff(self, keys, *args):
        """Returns the difference of sets specified by ``keys``"""
        args = list_or_args(keys, args)
        return await self.execute_command('SDIFF', *args)

    async def sdiffstore(self, dest, keys, *args):
        """
        Stores the difference of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.
        """
        args = list_or_args(keys, args)
        return await self.execute_command('SDIFFSTORE', dest, *args)

    async def sinter(self, keys, *args):
        """Returns the intersection of sets specified by ``keys``"""
        args = list_or_args(keys, args)
        return await self.execute_command('SINTER', *args)

    async def sinterstore(self, dest, keys, *args):
        """
        Stores the intersection of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.
        """
        args = list_or_args(keys, args)
        return await self.execute_command('SINTERSTORE', dest, *args)

    async def sismember(self, name, value):
        """
        Returns a boolean indicating if ``value`` is a member of set ``name``
        """
        return await self.execute_command('SISMEMBER', name, value)

    async def smembers(self, name):
        """Returns all members of the set ``name``"""
        return await self.execute_command('SMEMBERS', name)

    async def smove(self, src, dst, value):
        """Moves ``value`` from set ``src`` to set ``dst`` atomically"""
        return await self.execute_command('SMOVE', src, dst, value)

    async def spop(self, name, count=None):
        """
        Removes and returns a random member of set ``name``
        ``count`` should be type of int and default set to 1.
        If ``count`` is supplied, pops a list of ``count`` random
        members of set ``name``
        """
        if count and isinstance(count, int):
            return await self.execute_command('SPOP', name, count)
        else:
            return await self.execute_command('SPOP', name)

    async def srandmember(self, name, number=None):
        """
        If ``number`` is None, returns a random member of set ``name``.

        If ``number`` is supplied, returns a list of ``number`` random
        memebers of set ``name``. Note this is only available when running
        Redis 2.6+.
        """
        args = number and [number] or []
        return await self.execute_command('SRANDMEMBER', name, *args)

    async def srem(self, name, *values):
        """Removes ``values`` from set ``name``"""
        return await self.execute_command('SREM', name, *values)

    async def sunion(self, keys, *args):
        """Returns the union of sets specified by ``keys``"""
        args = list_or_args(keys, args)
        return await self.execute_command('SUNION', *args)

    async def sunionstore(self, dest, keys, *args):
        """
        Stores the union of sets specified by ``keys`` into a new
        set named ``dest``.  Returns the number of keys in the new set.
        """
        args = list_or_args(keys, args)
        return await self.execute_command('SUNIONSTORE', dest, *args)

    async def sscan(self, name, cursor=0, match=None, count=None):
        """
        Incrementally returns lists of elements in a set. Also returns a
        cursor pointing to the scan position.

        ``match`` is for filtering the keys by pattern

        ``count`` is for hint the minimum number of returns
        """
        pieces = [name, cursor]
        if match is not None:
            pieces.extend([b('MATCH'), match])
        if count is not None:
            pieces.extend([b('COUNT'), count])
        return await self.execute_command('SSCAN', *pieces) class ClusterSortedSetCommandMixin(SortedSetCommandMixin):

    RESULT_CALLBACKS = {
        'ZSCAN': first_key
    } class SortedSetCommandMixin:

    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict(
            'ZADD ZCARD ZLEXCOUNT '
            'ZREM ZREMRANGEBYLEX '
            'ZREMRANGEBYRANK '
            'ZREMRANGEBYSCORE', int
        ),
        string_keys_to_dict('ZSCORE ZINCRBY', float_or_none),
        string_keys_to_dict(
            'ZRANGE ZRANGEBYSCORE ZREVRANGE ZREVRANGEBYSCORE',
            zset_score_pairs
        ),
        string_keys_to_dict('ZRANK ZREVRANK', int_or_none),
        {
            'ZSCAN': parse_zscan,
        }
    )

    async def zadd(self, name, *args, **kwargs):
        """
        Set any number of score, element-name pairs to the key ``name``. Pairs
        can be specified in two ways:

        As *args, in the form of: score1, name1, score2, name2, ...
        or as **kwargs, in the form of: name1=score1, name2=score2, ...

        The following example would add four values to the 'my-key' key:
        redis.zadd('my-key', 1.1, 'name1', 2.2, 'name2', name3=3.3, name4=4.4)
        """
        pieces = []
        if args:
            if len(args) % 2 != 0:
                raise RedisError("ZADD requires an equal number of "
                                 "values and scores")
            pieces.extend(args)
        for pair in iteritems(kwargs):
            pieces.append(pair[1])
            pieces.append(pair[0])
        return await self.execute_command('ZADD', name, *pieces)

    async def zaddoption(self, name, option=None, *args, **kwargs):
        """
        Differs from zadd in that you can set either 'XX' or 'NX' option as
        described here: https://redis.io/commands/zadd. Only for Redis 3.0.2 or
        later.

        The following example would add four values to the 'my-key' key:
        redis.zaddoption('my-key', 'XX', 1.1, 'name1', 2.2, 'name2', name3=3.3, name4=4.4)
        redis.zaddoption('my-key', 'NX CH', name1=2.2)
        """
        if not option:
            raise RedisError("ZADDOPTION must take options")
        options = set(opt.upper() for opt in option.split())
        if options - VALID_ZADD_OPTIONS:
            raise RedisError("ZADD only takes XX, NX, CH, or INCR")
        if 'NX' in options and 'XX' in options:
            raise RedisError("ZADD only takes one of XX or NX")
        pieces = list(options)
        members = []
        if args:
            if len(args) % 2 != 0:
                raise RedisError("ZADD requires an equal number of "
                                 "values and scores")
            members.extend(args)
        for pair in iteritems(kwargs):
            members.append(pair[1])
            members.append(pair[0])
        if 'INCR' in options and len(members) != 2:
            raise RedisError("ZADD with INCR only takes one score-name pair")
        return await self.execute_command('ZADD', name, *pieces, *members)

    async def zcard(self, name):
        """Returns the number of elements in the sorted set ``name``"""
        return await self.execute_command('ZCARD', name)

    async def zcount(self, name, min, max):
        """
        Returns the number of elements in the sorted set at key ``name`` with
        a score between ``min`` and ``max``.
        """
        return await self.execute_command('ZCOUNT', name, min, max)

    async def zincrby(self, name, value, amount=1):
        """
        Increments the score of ``value`` in sorted set ``name`` by ``amount``
        """
        return await self.execute_command('ZINCRBY', name, amount, value)

    async def zinterstore(self, dest, keys, aggregate=None):
        """
        Intersects multiple sorted sets specified by ``keys`` into
        a new sorted set, ``dest``. Scores in the destination will be
        aggregated based on the ``aggregate``, or SUM if none is provided.
        """
        return await self._zaggregate('ZINTERSTORE', dest, keys, aggregate)

    async def zlexcount(self, name, min, max):
        """
        Returns the number of items in the sorted set ``name`` between the
        lexicographical range ``min`` and ``max``.
        """
        return await self.execute_command('ZLEXCOUNT', name, min, max)

    async def zrange(self, name, start, end, desc=False, withscores=False,
                     score_cast_func=float):
        """
        Returns a range of values from sorted set ``name`` between
        ``start`` and ``end`` sorted in ascending order.

        ``start`` and ``end`` can be negative, indicating the end of the range.

        ``desc`` a boolean indicating whether to sort the results descendingly

        ``withscores`` indicates to return the scores along with the values.
        The return type is a list of (value, score) pairs

        ``score_cast_func`` a callable used to cast the score return value
        """
        if desc:
            return await self.zrevrange(name, start, end, withscores,
                                        score_cast_func)
        pieces = ['ZRANGE', name, start, end]
        if withscores:
            pieces.append(b('WITHSCORES'))
        options = {
            'withscores': withscores,
            'score_cast_func': score_cast_func
        }
        return await self.execute_command(*pieces, **options)

    async def zrangebylex(self, name, min, max, start=None, num=None):
        """
        Returns the lexicographical range of values from sorted set ``name``
        between ``min`` and ``max``.

        If ``start`` and ``num`` are specified, then return a slice of the
        range.
        """
        if (start is not None and num is None) or \
                (num is not None and start is None):
            raise RedisError("``start`` and ``num`` must both be specified")
        pieces = ['ZRANGEBYLEX', name, min, max]
        if start is not None and num is not None:
            pieces.extend([b('LIMIT'), start, num])
        return await self.execute_command(*pieces)

    async def zrevrangebylex(self, name, max, min, start=None, num=None):
        """
        Returns the reversed lexicographical range of values from sorted set
        ``name`` between ``max`` and ``min``.

        If ``start`` and ``num`` are specified, then return a slice of the
        range.
        """
        if (start is not None and num is None) or \
                (num is not None and start is None):
            raise RedisError("``start`` and ``num`` must both be specified")
        pieces = ['ZREVRANGEBYLEX', name, max, min]
        if start is not None and num is not None:
            pieces.extend([b('LIMIT'), start, num])
        return await self.execute_command(*pieces)

    async def zrangebyscore(self, name, min, max, start=None, num=None,
                            withscores=False, score_cast_func=float):
        """
        Returns a range of values from the sorted set ``name`` with scores
        between ``min`` and ``max``.

        If ``start`` and ``num`` are specified, then return a slice
        of the range.

        ``withscores`` indicates to return the scores along with the values.
        The return type is a list of (value, score) pairs

        `score_cast_func`` a callable used to cast the score return value
        """
        if (start is not None and num is None) or \
                (num is not None and start is None):
            raise RedisError("``start`` and ``num`` must both be specified")
        pieces = ['ZRANGEBYSCORE', name, min, max]
        if start is not None and num is not None:
            pieces.extend([b('LIMIT'), start, num])
        if withscores:
            pieces.append(b('WITHSCORES'))
        options = {
            'withscores': withscores,
            'score_cast_func': score_cast_func
        }
        return await self.execute_command(*pieces, **options)

    async def zrank(self, name, value):
        """
        Returns a 0-based value indicating the rank of ``value`` in sorted set
        ``name``
        """
        return await self.execute_command('ZRANK', name, value)

    async def zrem(self, name, *values):
        """Removes member ``values`` from sorted set ``name``"""
        return await self.execute_command('ZREM', name, *values)

    async def zremrangebylex(self, name, min, max):
        """
        Removes all elements in the sorted set ``name`` between the
        lexicographical range specified by ``min`` and ``max``.

        Returns the number of elements removed.
        """
        return await self.execute_command('ZREMRANGEBYLEX', name, min, max)

    async def zremrangebyrank(self, name, min, max):
        """
        Removes all elements in the sorted set ``name`` with ranks between
        ``min`` and ``max``. Values are 0-based, ordered from smallest score
        to largest. Values can be negative indicating the highest scores.
        Returns the number of elements removed
        """
        return await self.execute_command('ZREMRANGEBYRANK', name, min, max)

    async def zremrangebyscore(self, name, min, max):
        """
        Removes all elements in the sorted set ``name`` with scores
        between ``min`` and ``max``. Returns the number of elements removed.
        """
        return await self.execute_command('ZREMRANGEBYSCORE', name, min, max)

    async def zrevrange(self, name, start, end, withscores=False,
                        score_cast_func=float):
        """
        Returns a range of values from sorted set ``name`` between
        ``start`` and ``end`` sorted in descending order.

        ``start`` and ``end`` can be negative, indicating the end of the range.

        ``withscores`` indicates to return the scores along with the values
        The return type is a list of (value, score) pairs

        ``score_cast_func`` a callable used to cast the score return value
        """
        pieces = ['ZREVRANGE', name, start, end]
        if withscores:
            pieces.append(b('WITHSCORES'))
        options = {
            'withscores': withscores,
            'score_cast_func': score_cast_func
        }
        return await self.execute_command(*pieces, **options)

    async def zrevrangebyscore(self, name, max, min, start=None, num=None,
                               withscores=False, score_cast_func=float):
        """
        Returns a range of values from the sorted set ``name`` with scores
        between ``min`` and ``max`` in descending order.

        If ``start`` and ``num`` are specified, then return a slice
        of the range.

        ``withscores`` indicates to return the scores along with the values.
        The return type is a list of (value, score) pairs

        ``score_cast_func`` a callable used to cast the score return value
        """
        if (start is not None and num is None) or \
                (num is not None and start is None):
            raise RedisError("``start`` and ``num`` must both be specified")
        pieces = ['ZREVRANGEBYSCORE', name, max, min]
        if start is not None and num is not None:
            pieces.extend([b('LIMIT'), start, num])
        if withscores:
            pieces.append(b('WITHSCORES'))
        options = {
            'withscores': withscores,
            'score_cast_func': score_cast_func
        }
        return await self.execute_command(*pieces, **options)

    async def zrevrank(self, name, value):
        """
        Returns a 0-based value indicating the descending rank of
        ``value`` in sorted set ``name``
        """
        return await self.execute_command('ZREVRANK', name, value)

    async def zscore(self, name, value):
        "Return the score of element ``value`` in sorted set ``name``"
        return await self.execute_command('ZSCORE', name, value)

    async def zunionstore(self, dest, keys, aggregate=None):
        """
        Performs Union on multiple sorted sets specified by ``keys`` into
        a new sorted set, ``dest``. Scores in the destination will be
        aggregated based on the ``aggregate``, or SUM if none is provided.
        """
        return await self._zaggregate('ZUNIONSTORE', dest, keys, aggregate)

    async def _zaggregate(self, command, dest, keys, aggregate=None):
        pieces = [command, dest, len(keys)]
        if isinstance(keys, dict):
            keys, weights = iterkeys(keys), itervalues(keys)
        else:
            weights = None
        pieces.extend(keys)
        if weights:
            pieces.append(b('WEIGHTS'))
            pieces.extend(weights)
        if aggregate:
            pieces.append(b('AGGREGATE'))
            pieces.append(aggregate)
        return await self.execute_command(*pieces)

    async def zscan(self, name, cursor=0, match=None, count=None,
                    score_cast_func=float):
        """
        Incrementally returns lists of elements in a sorted set. Also returns
        a cursor pointing to the scan position.

        ``match`` allows for filtering the keys by pattern

        ``count`` allows for hint the minimum number of returns

        ``score_cast_func`` a callable used to cast the score return value
        """
        pieces = [name, cursor]
        if match is not None:
            pieces.extend([b('MATCH'), match])
        if count is not None:
            pieces.extend([b('COUNT'), count])
        options = {'score_cast_func': score_cast_func}
        return await self.execute_command('ZSCAN', *pieces, **options) class StreamsCommandMixin:
    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict('XREVRANGE XRANGE', stream_list),
        string_keys_to_dict('XREAD XREADGROUP', multi_stream_list),
        {
            'XINFO GROUPS': list_of_pairs_to_dict,
            'XINFO STREAM': parse_xinfo_stream,
            'XINFO CONSUMERS': list_of_pairs_to_dict,
            'XGROUP SETID': bool_ok,
            'XGROUP CREATE': bool_ok
        }
    )

    async def xadd(self, name: str, entry: dict,
                   max_len=None, stream_id='*',
                   approximate=True) -> str:
        """
        Appends the specified stream entry to the stream at the specified key.
        If the key does not exist, as a side effect of running
        this command the key is created with a stream value.
        Available since 5.0.0.
        Time complexity: O(log(N)) with N being the number of items already into the stream.

        :param name: name of the stream
        :param entry: key-values to be appended to the stream
        :param max_len: max length of the stream
        length will not be limited max_len is set to None
        notice: max_len should be int greater than 0,
        if set to 0 or negative, the stream length will not be limited

        :param stream_id: id of the options appended to the stream.
        The XADD command will auto-generate a unique id for you
        if the id argument specified is the * character.
        ID are specified by two numbers separated by a "-" character

        :param approximate: whether redis will limit
        the stream with given max length exactly, if set to True,
        there will be a few tens of entries more,
        but never less than 1000 items

        :return: id auto generated or the specified id given.
        notice: specified id without "-" character will be completed like "id-0"
        """
        pieces = []
        if max_len is not None:
            if not isinstance(max_len, int) or max_len < 1:
                raise RedisError("XADD maxlen must be a positive integer")
            pieces.append('MAXLEN')
            if approximate:
                pieces.append('~')
            pieces.append(str(max_len))
        pieces.append(stream_id)
        for kv in entry.items():
            pieces.extend(list(kv))
        return await self.execute_command('XADD', name, *pieces)

    async def xlen(self, name: str) -> int:
        """
        Returns the number of elements in a given stream.
        """
        return await self.execute_command('XLEN', name)

    async def xrange(self, name: str, start='-', end='+', count=None) -> list:
        """
        Read stream values within an interval.

        Available since 5.0.0.
        Time complexity: O(log(N)+M) with N being the number of elements in the stream and M the number
        of elements being returned. If M is constant (e.g. always asking for the first 10 elements with COUNT),
        you can consider it O(log(N)).

        :param name: name of the stream.
        :param start: first stream ID. defaults to '-',
               meaning the earliest available.
        :param end: last stream ID. defaults to '+',
                meaning the latest available.
        :param count: if set, only return this many items, beginning with the
               earliest available.
        :return list of (stream_id, entry(k-v pair))
        """

        pieces = [start, end]
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise RedisError("XRANGE count must be a positive integer")
            pieces.append("COUNT")
            pieces.append(str(count))
        return await self.execute_command('XRANGE', name, *pieces)

    async def xrevrange(self, name: str, start='+', end='-', count=None) -> list:
        """
        Read stream values within an interval, in reverse order.

        Available since 5.0.0.
        Time complexity: O(log(N)+M) with N being the number of elements in the stream and M the number
        of elements being returned. If M is constant (e.g. always asking for the first 10 elements with COUNT),
        you can consider it O(log(N)).

        :param name: name of the stream
        :param start: first stream ID. defaults to '+',
               meaning the latest available.
        :param end: last stream ID. defaults to '-',
                meaning the earliest available.
        :param count: if set, only return this many items, beginning with the
               latest available.

        """
        pieces = [start, end]
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise RedisError("XREVRANGE count must be a positive integer")
            pieces.append("COUNT")
            pieces.append(str(count))
        return await self.execute_command('XREVRANGE', name, *pieces)

    async def xread(self, count=None, block=None, **streams) -> dict:
        """
        Available since 5.0.0.

        Time complexity:
        For each stream mentioned: O(log(N)+M) with N being the number
        of elements in the stream and M the number of elements being returned.
        If M is constant (e.g. always asking for the first 10 elements with COUNT),
        you can consider it O(log(N)). On the other side, XADD will pay the O(N)
        time in order to serve the N clients blocked on the stream getting new data.

        Read data from one or multiple streams,
        only returning entries with an ID greater
        than the last received ID reported by the caller.

        :param count: int, if set, only return this many items, beginning with the
               earliest available.
        :param block: int, milliseconds we want to block before timing out,
                if the BLOCK option is not used, the command is synchronous
        :param streams: stream_name - stream_id mapping
        :return dict like {stream_name: [(stream_id: entry), ...]}
        """
        pieces = []
        if block is not None:
            if not isinstance(block, int) or block < 0:
                raise RedisError("XREAD block must be a positive integer")
            pieces.append("BLOCK")
            pieces.append(str(block))
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise RedisError("XREAD count must be a positive integer")
            pieces.append("COUNT")
            pieces.append(str(count))
        pieces.append("STREAMS")
        ids = []
        for partial_stream in streams.items():
            pieces.append(partial_stream[0])
            ids.append(partial_stream[1])
        pieces.extend(ids)
        return await self.execute_command('XREAD', *pieces)

    async def xreadgroup(self, group: str, consumer_id: str,
                         count=None, block=None, **streams):
        """
        Available since 5.0.0.

        Time complexity:
        For each stream mentioned: O(log(N)+M) with N being the number of elements
        in the stream and M the number of elements being returned.
        If M is constant (e.g. always asking for the first 10 elements with COUNT),
        you can consider it O(log(N)). On the other side,
        XADD will pay the O(N) time in order to serve
        the N clients blocked on the stream getting new data.

        Read data from one or multiple streams via the consumer group,
        only returning entries with an ID greater
        than the last received ID reported by the caller.

        :param group: the name of the consumer group
        :param consumer_id: the name of the consumer that is attempting to read
        :param count: int, if set, only return this many items, beginning with the
               earliest available.
        :param block: int, milliseconds we want to block before timing out,
                if the BLOCK option is not used, the command is synchronous
        :param streams: stream_name - stream_id mapping
        :return dict like {stream_name: [(stream_id: entry), ...]}
        """
        pieces = ['GROUP', group, consumer_id]
        if block is not None:
            if not isinstance(block, int) or block < 1:
                raise RedisError("XREAD block must be a positive integer")
            pieces.append("BLOCK")
            pieces.append(str(block))
        if count is not None:
            if not isinstance(count, int) or count < 1:
                raise RedisError("XREAD count must be a positive integer")
            pieces.append("COUNT")
            pieces.append(str(count))
        pieces.append("STREAMS")
        ids = []
        for partial_stream in streams.items():
            pieces.append(partial_stream[0])
            ids.append(partial_stream[1])
        pieces.extend(ids)
        return await self.execute_command('XREADGROUP', *pieces)

    async def xpending(self, name: str, group: str,
                       start='-', end='+', count=None, consumer=None) -> list:
        """
        Available since 5.0.0.

        Time complexity:
        O(log(N)+M) with N being the number of elements in the consumer
        group pending entries list, and M the number of elements being returned.
        When the command returns just the summary it runs in O(1)
        time assuming the list of consumers is small,
        otherwise there is additional O(N) time needed to iterate every consumer.

        Fetching data from a stream via a consumer group,
        and not acknowledging such data,
        has the effect of creating pending entries.
        The XPENDING command is the interface to inspect the list of pending messages.

        :param name: name of the stream
        :param group: name of the consumer group
        :param start: first stream ID. defaults to '-',
               meaning the earliest available.
        :param end: last stream ID. defaults to '+',
                meaning the latest available.
        :param count: int, number of entries
                [NOTICE] only when count is set to int,
                start & end options will have effect
                and detail of pending entries will be returned
        :param consumer: str, consumer of the stream in the group
                [NOTICE] only when count is set to int,
                this option can be appended to
                query pending entries of given consumer
        """
        pieces = [name, group]
        if count is not None:
            pieces.extend([start, end, count])
            if consumer is not None:
                pieces.append(str(consumer))
        # todo: may there be a parse function
        return await self.execute_command('XPENDING', *pieces)

    async def xtrim(self, name: str, max_len: int, approximate=True) -> int:
        """
        [NOTICE] Not officially released yet

        XTRIM is designed to accept different trimming strategies,
        even if currently only MAXLEN is implemented.

        :param name: name of the stream
        :param max_len: max length of the stream after being trimmed
        :param approximate: whether redis will limit
        the stream with given max length exactly, if set to True,
        there will be a few tens of entries more,
        but never less than 1000 items:

        :return: number of entries trimmed
        """
        pieces = ['MAXLEN']
        if approximate:
            pieces.append('~')
        pieces.append(max_len)
        return await self.execute_command('XTRIM', name, *pieces)

    async def xdel(self, name: str, stream_id: str) -> int:
        """
        [NOTICE] Not officially released yet
        [NOTICE] In the current implementation, memory is not
        really reclaimed until a macro node is completely empty,
        so you should not abuse this feature.

        remove items from the middle of a stream, just by ID.

        :param name: name of the stream
        :param stream_id: id of the options appended to the stream.
        """
        return await self.execute_command('XDEL', name, stream_id)

    async def xinfo_consumers(self, name: str, group: str) -> list:
        """
        [NOTICE] Not officially released yet

        XINFO command is an observability interface that can be used
        with sub-commands in order to get information
        about streams or consumer groups.

        :param name: name of the stream
        :param group: name of the consumer group
        """
        return await self.execute_command('XINFO CONSUMERS', name, group)

    async def xinfo_groups(self, name: str) -> list:
        """
        [NOTICE] Not officially released yet

        XINFO command is an observability interface that can be used
        with sub-commands in order to get information
        about streams or consumer groups.

        :param name: name of the stream
        """
        return await self.execute_command('XINFO GROUPS', name)

    async def xinfo_stream(self, name: str) -> dict:
        """
        [NOTICE] Not officially released yet

        XINFO command is an observability interface that can be used
        with sub-commands in order to get information
        about streams or consumer groups.

        :param name: name of the stream
        """
        return await self.execute_command('XINFO STREAM', name)

    async def xack(self, name: str, group: str, stream_id: str) -> int:
        """
        [NOTICE] Not officially released yet

        XACK is the command that allows a consumer to mark a pending message as correctly processed.

        :param name: name of the stream
        :param group: name of the consumer group
        :param stream_id: id of the entry the consumer wants to mark
        :return: number of entry marked
        """
        return await self.execute_command('XACK', name, group, stream_id)

    async def xclaim(self, name: str, group: str, consumer: str, min_idle_time: int, *stream_ids):
        """
        [NOTICE] Not officially released yet

        Gets ownership of one or multiple messages in the Pending Entries List of a given stream consumer group.

        :param name: name of the stream
        :param group: name of the consumer group
        :param consumer: name of the consumer
        :param min_idle_time: ms
            If the message ID (among the specified ones) exists, and its idle time greater
            or equal to min_idle_time, then the message new owner
            becomes the specified <consumer>. If the minimum idle time specified
            is zero, messages are claimed regardless of their idle time.
        :param stream_ids:
        """
        return await self.execute_command('XCLAIM', name, group, consumer, min_idle_time, *stream_ids)

    async def xgroup_create(self, name: str, group: str, stream_id='$') -> bool:
        """
        [NOTICE] Not officially released yet
        XGROUP is used in order to create, destroy and manage consumer groups.
        :param name: name of the stream
        :param group: name of the consumer group
        :param stream_id:
            If we provide $ as we did, then only new messages arriving
            in the stream from now on will be provided to the consumers in the group.
            If we specify 0 instead the consumer group will consume all the messages
            in the stream history to start with.
            Of course, you can specify any other valid ID
        """
        return await self.execute_command('XGROUP CREATE', name, group, stream_id)

    async def xgroup_set_id(self, name: str, group: str, stream_id: str) -> bool:
        """
        [NOTICE] Not officially released yet
        :param name: name of the stream
        :param group: name of the consumer group
        :param stream_id:
            If we provide $ as we did, then only new messages arriving
            in the stream from now on will be provided to the consumers in the group.
            If we specify 0 instead the consumer group will consume all the messages
            in the stream history to start with.
            Of course, you can specify any other valid ID
        """
        return await self.execute_command('XGROUP SETID', name, group, stream_id)

    async def xgroup_destroy(self, name: str, group: str) -> int:
        """
        [NOTICE] Not officially released yet
        XGROUP is used in order to create, destroy and manage consumer groups.
        :param name: name of the stream
        :param group: name of the consumer group
        """
        return await self.execute_command('XGROUP DESTROY', name, group)

    async def xgroup_del_consumer(self, name: str, group: str, consumer: str) -> int:
        """
        [NOTICE] Not officially released yet
        XGROUP is used in order to create, destroy and manage consumer groups.
        :param name: name of the stream
        :param group: name of the consumer group
        :param consumer: name of the consumer
        """
        return await self.execute_command('XGROUP DELCONSUMER', name, group, consumer) class ClusterStringsCommandMixin(StringsCommandMixin):

    NODES_FLAGS = {
        'BITOP': NodeFlag.BLOCKED
    }

    async def mget(self, keys, *args):
        """
        Returns a list of values ordered identically to ``keys``

        Cluster impl:
            Itterate all keys and send GET for each key.
            This will go alot slower than a normal mget call in StrictRedis.

            Operation is no longer atomic.
        """
        res = list()
        for arg in list_or_args(keys, args):
            res.append(await self.get(arg))
        return res

    async def mset(self, *args, **kwargs):
        """
        Sets key/values based on a mapping. Mapping can be supplied as a single
        dictionary argument or as kwargs.

        Cluster impl:
            Itterate over all items and do SET on each (k,v) pair

            Operation is no longer atomic.
        """
        if args:
            if len(args) != 1 or not isinstance(args[0], dict):
                raise RedisError('MSET requires **kwargs or a single dict arg')
            kwargs.update(args[0])

        for pair in iteritems(kwargs):
            await self.set(pair[0], pair[1])

        return True

    async def msetnx(self, *args, **kwargs):
        """
        Sets key/values based on a mapping if none of the keys are already set.
        Mapping can be supplied as a single dictionary argument or as kwargs.
        Returns a boolean indicating if the operation was successful.

        Clutser impl:
            Itterate over all items and do GET to determine if all keys do not exists.
            If true then call mset() on all keys.
        """
        if args:
            if len(args) != 1 or not isinstance(args[0], dict):
                raise RedisError('MSETNX requires **kwargs or a single dict arg')
            kwargs.update(args[0])

        # Itterate over all items and fail fast if one value is True.
        for k, _ in kwargs.items():
            if await self.get(k):
                return False

        return await self.mset(**kwargs) class StringsCommandMixin:
    RESPONSE_CALLBACKS = dict_merge(
        string_keys_to_dict(
            'MSETNX PSETEX SETEX SETNX',
            bool
        ),
        string_keys_to_dict(
            'BITCOUNT BITPOS DECRBY GETBIT INCRBY '
            'STRLEN SETBIT', int
        ),
        {
            'INCRBYFLOAT': float,
            'MSET': bool_ok,
            'SET': lambda r: r and nativestr(r) == 'OK',
        }
    )

    async def append(self, key, value):
        """
        Appends the string ``value`` to the value at ``key``. If ``key``
        doesn't already exist, create it with a value of ``value``.
        Returns the new length of the value at ``key``.
        """
        return await self.execute_command('APPEND', key, value)

    async def bitcount(self, key, start=None, end=None):
        """
        Returns the count of set bits in the value of ``key``.  Optional
        ``start`` and ``end`` paramaters indicate which bytes to consider
        """
        params = [key]
        if start is not None and end is not None:
            params.append(start)
            params.append(end)
        elif (start is not None and end is None) or \
                (end is not None and start is None):
            raise RedisError("Both start and end must be specified")
        return await self.execute_command('BITCOUNT', *params)

    async def bitop(self, operation, dest, *keys):
        """
        Perform a bitwise operation using ``operation`` between ``keys`` and
        store the result in ``dest``.
        """
        return await self.execute_command('BITOP', operation, dest, *keys)

    async def bitpos(self, key, bit, start=None, end=None):
        """
        Return the position of the first bit set to 1 or 0 in a string.
        ``start`` and ``end`` difines search range. The range is interpreted
        as a range of bytes and not a range of bits, so start=0 and end=2
        means to look at the first three bytes.
        """
        if bit not in (0, 1):
            raise RedisError('bit must be 0 or 1')
        params = [key, bit]

        start is not None and params.append(start)

        if start is not None and end is not None:
            params.append(end)
        elif start is None and end is not None:
            raise RedisError("start argument is not set, "
                             "when end is specified")
        return await self.execute_command('BITPOS', *params)

    def bitfield(self, key):
        return BitField(self, key)

    async def decr(self, name, amount=1):
        """
        Decrements the value of ``key`` by ``amount``.  If no key exists,
        the value will be initialized as 0 - ``amount``
        """
        return await self.execute_command('DECRBY', name, amount)

    async def get(self, name):
        """
        Return the value at key ``name``, or None if the key doesn't exist
        """
        return await self.execute_command('GET', name)

    async def getbit(self, name, offset):
        "Returns a boolean indicating the value of ``offset`` in ``name``"
        return await self.execute_command('GETBIT', name, offset)

    async def getrange(self, key, start, end):
        """
        Returns the substring of the string value stored at ``key``,
        determined by the offsets ``start`` and ``end`` (both are inclusive)
        """
        return await self.execute_command('GETRANGE', key, start, end)

    async def getset(self, name, value):
        """
        Sets the value at key ``name`` to ``value``
        and returns the old value at key ``name`` atomically.
        """
        return await self.execute_command('GETSET', name, value)

    async def incr(self, name, amount=1):
        """
        Increments the value of ``key`` by ``amount``.  If no key exists,
        the value will be initialized as ``amount``
        """
        return await self.execute_command('INCRBY', name, amount)

    async def incrby(self, name, amount=1):
        """
        Increments the value of ``key`` by ``amount``.  If no key exists,
        the value will be initialized as ``amount``
        """

        # An alias for ``incr()``, because it is already implemented
        # as INCRBY redis command.
        return await self.incr(name, amount)

    async def incrbyfloat(self, name, amount=1.0):
        """
        Increments the value at key ``name`` by floating ``amount``.
        If no key exists, the value will be initialized as ``amount``
        """
        return await self.execute_command('INCRBYFLOAT', name, amount)

    async def mget(self, keys, *args):
        """
        Returns a list of values ordered identically to ``keys``
        """
        args = list_or_args(keys, args)
        return await self.execute_command('MGET', *args)

    async def mset(self, *args, **kwargs):
        """
        Sets key/values based on a mapping. Mapping can be supplied as a single
        dictionary argument or as kwargs.
        """
        if args:
            if len(args) != 1 or not isinstance(args[0], dict):
                raise RedisError('MSET requires **kwargs or a single dict arg')
            kwargs.update(args[0])
        items = []
        for pair in iteritems(kwargs):
            items.extend(pair)
        return await self.execute_command('MSET', *items)

    async def msetnx(self, *args, **kwargs):
        """
        Sets key/values based on a mapping if none of the keys are already set.
        Mapping can be supplied as a single dictionary argument or as kwargs.
        Returns a boolean indicating if the operation was successful.
        """
        if args:
            if len(args) != 1 or not isinstance(args[0], dict):
                raise RedisError('MSETNX requires **kwargs or a single '
                                 'dict arg')
            kwargs.update(args[0])
        items = []
        for pair in iteritems(kwargs):
            items.extend(pair)
        return await self.execute_command('MSETNX', *items)

    async def psetex(self, name, time_ms, value):
        """
        Set the value of key ``name`` to ``value`` that expires in ``time_ms``
        milliseconds. ``time_ms`` can be represented by an integer or a Python
        timedelta object
        """
        if isinstance(time_ms, datetime.timedelta):
            ms = int(time_ms.microseconds / 1000)
            time_ms = (time_ms.seconds + time_ms.days * 24 * 3600) * 1000 + ms
        return await self.execute_command('PSETEX', name, time_ms, value)

    async def set(self, name, value, ex=None, px=None, nx=False, xx=False):
        """
        Set the value at key ``name`` to ``value``

        ``ex`` sets an expire flag on key ``name`` for ``ex`` seconds.

        ``px`` sets an expire flag on key ``name`` for ``px`` milliseconds.

        ``nx`` if set to True, set the value at key ``name`` to ``value`` if it
            does not already exist.

        ``xx`` if set to True, set the value at key ``name`` to ``value`` if it
            already exists.
        """
        pieces = [name, value]
        if ex is not None:
            pieces.append('EX')
            if isinstance(ex, datetime.timedelta):
                ex = ex.seconds + ex.days * 24 * 3600
            pieces.append(ex)
        if px is not None:
            pieces.append('PX')
            if isinstance(px, datetime.timedelta):
                ms = int(px.microseconds / 1000)
                px = (px.seconds + px.days * 24 * 3600) * 1000 + ms
            pieces.append(px)

        if nx:
            pieces.append('NX')
        if xx:
            pieces.append('XX')
        return await self.execute_command('SET', *pieces)

    async def setbit(self, name, offset, value):
        """
        Flag the ``offset`` in ``name`` as ``value``. Returns a boolean
        indicating the previous value of ``offset``.
        """
        value = value and 1 or 0
        return await self.execute_command('SETBIT', name, offset, value)

    async def setex(self, name, time, value):
        """
        Set the value of key ``name`` to ``value`` that expires in ``time``
        seconds. ``time`` can be represented by an integer or a Python
        timedelta object.
        """
        if isinstance(time, datetime.timedelta):
            time = time.seconds + time.days * 24 * 3600
        return await self.execute_command('SETEX', name, time, value)

    async def setnx(self, name, value):
        """
        Sets the value of key ``name`` to ``value`` if key doesn't exist
        """
        return await self.execute_command('SETNX', name, value)

    async def setrange(self, name, offset, value):
        """
        Overwrite bytes in the value of ``name`` starting at ``offset`` with
        ``value``. If ``offset`` plus the length of ``value`` exceeds the
        length of the original value, the new value will be larger than before.
        If ``offset`` exceeds the length of the original value, null bytes
        will be used to pad between the end of the previous value and the start
        of what's being injected.

        Returns the length of the new string.
        """
        return await self.execute_command('SETRANGE', name, offset, value)

    async def strlen(self, name):
        """Returns the number of bytes stored in the value of ``name``"""
        return await self.execute_command('STRLEN', name)

    async def substr(self, name, start, end=-1):
        """
        Returns a substring of the string at key ``name``. ``start`` and ``end``
        are 0-based integers specifying the portion of the string to return.
        """
        return await self.execute_command('SUBSTR', name, start, end) class ClusterTransactionCommandMixin(TransactionCommandMixin):

    async def transaction(self, func, *watches, **kwargs):
        """
        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.

        cluster transaction can only be run with commands in the same node,
        otherwise error will be raised.
        """
        shard_hint = kwargs.pop('shard_hint', None)
        value_from_callable = kwargs.pop('value_from_callable', False)
        watch_delay = kwargs.pop('watch_delay', None)
        async with await self.pipeline(True, shard_hint, watches=watches) as pipe:
            while True:
                try:
                    func_value = await func(pipe)
                    exec_value = await pipe.execute()
                    return func_value if value_from_callable else exec_value
                except WatchError:
                    if watch_delay is not None and watch_delay > 0:
                        await asyncio.sleep(
                            watch_delay,
                            loop=self.connection_pool.loop
                        )
                    continue class TransactionCommandMixin:

    RESPONSE_CALLBACKS = string_keys_to_dict(
            'WATCH UNWATCH',
            bool_ok
        )

    async def transaction(self, func, *watches, **kwargs):
        """
        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.
        """
        shard_hint = kwargs.pop('shard_hint', None)
        value_from_callable = kwargs.pop('value_from_callable', False)
        watch_delay = kwargs.pop('watch_delay', None)
        async with await self.pipeline(True, shard_hint) as pipe:
            while True:
                try:
                    if watches:
                        await pipe.watch(*watches)
                    func_value = await func(pipe)
                    exec_value = await pipe.execute()
                    return func_value if value_from_callable else exec_value
                except WatchError:
                    if watch_delay is not None and watch_delay > 0:
                        await asyncio.sleep(
                            watch_delay,
                            loop=self.connection_pool.loop
                        )
                    continue

    async def watch(self, *names):
        """
        Watches the values at keys ``names``, or None if the key doesn't exist
        """
        warnings.warn(DeprecationWarning('Call WATCH from a Pipeline object'))

    async def unwatch(self):
        """
        Unwatches the value at key ``name``, or None of the key doesn't exist
        """
        warnings.warn(
            DeprecationWarning('Call UNWATCH from a Pipeline object')) class RedisSSLContext:
    def __init__(self, keyfile=None, certfile=None,
                 cert_reqs=None, ca_certs=None):
        self.keyfile = keyfile
        self.certfile = certfile
        if cert_reqs is None:
            self.cert_reqs = ssl.CERT_NONE
        elif isinstance(cert_reqs, str):
            CERT_REQS = {
                'none': ssl.CERT_NONE,
                'optional': ssl.CERT_OPTIONAL,
                'required': ssl.CERT_REQUIRED
            }
            if cert_reqs not in CERT_REQS:
                raise RedisError(
                    "Invalid SSL Certificate Requirements Flag: %s" %
                    cert_reqs)
            self.cert_reqs = CERT_REQS[cert_reqs]
        self.ca_certs = ca_certs
        self.context = None

    def get(self):
        if not self.keyfile:
            self.context = ssl.create_default_context(cafile=self.ca_certs)
        else:
            self.context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
            self.context.verify_mode = self.cert_reqs
            self.context.load_cert_chain(certfile=self.certfile,
                                         keyfile=self.keyfile)
            self.context.load_verify_locations(self.ca_certs)
        return self.context class UnixDomainSocketConnection(BaseConnection):
    description = "UnixDomainSocketConnection<path={path},db={db}>"

    def __init__(self, path='', password=None,
                 db=0, retry_on_timeout=False, stream_timeout=None, connect_timeout=None,
                 ssl_context=None, parser_class=DefaultParser, reader_read_size=65535,
                 encoding='utf-8', decode_responses=False, *, loop=None):
        super(UnixDomainSocketConnection, self).__init__(retry_on_timeout, stream_timeout,
                                                         parser_class, reader_read_size,
                                                         encoding, decode_responses,
                                                         loop=loop)
        self.path = path
        self.db = db
        self.password = password
        self.ssl_context = ssl_context
        self._connect_timeout = connect_timeout
        self._description_args = {
            'path': self.path,
            'db': self.db
        }

    async def _connect(self):
        reader, writer = await exec_with_timeout(
            asyncio.open_unix_connection(path=self.path,
                                         ssl=self.ssl_context,
                                         loop=self.loop),
            self._connect_timeout,
            loop=self.loop
        )
        self._reader = reader
        self._writer = writer
        await self.on_connect() class AskError(ResponseError):
    """
    src node: MIGRATING to dst node
        get > ASK error
        ask dst node > ASKING command
    dst node: IMPORTING from src node
        asking command only affects next command
        any op will be allowed after asking command
    """

    def __init__(self, resp):
        """should only redirect to master node"""
        self.args = (resp,)
        self.message = resp
        slot_id, new_node = resp.split(' ')
        host, port = new_node.rsplit(':', 1)
        self.slot_id = int(slot_id)
        self.node_addr = self.host, self.port = host, int(port) class BusyLoadingError(ConnectionError):
    pass class ClusterDownError(ClusterError, ResponseError):

    def __init__(self, resp):
        self.args = (resp,)
        self.message = resp class ClusterError(RedisError):
    pass class ConnectionError(RedisError):
    pass class MovedError(AskError):
    pass class RedisClusterException(Exception):
    pass class TimeoutError(RedisError):
    pass class TryAgainError(ResponseError):

    def __init__(self, *args, **kwargs):
        pass class ClusterConnectionPool(ConnectionPool):
    """Custom connection pool for rediscluster"""
    RedisClusterDefaultTimeout = None

    def __init__(self, startup_nodes=None, connection_class=ClusterConnection,
                 max_connections=None, max_connections_per_node=False, reinitialize_steps=None,
                 skip_full_coverage_check=False, nodemanager_follow_cluster=False, readonly=False,
                 max_idle_time=0, idle_check_interval=1,
                 **connection_kwargs):
        """
        :skip_full_coverage_check:
            Skips the check of cluster-require-full-coverage config, useful for clusters
            without the CONFIG command (like aws)
        :nodemanager_follow_cluster:
            The node manager will during initialization try the last set of nodes that
            it was operating on. This will allow the client to drift along side the cluster
            if the cluster nodes move around alot.
        """
        super(ClusterConnectionPool, self).__init__(connection_class=connection_class, max_connections=max_connections)

        # Special case to make from_url method compliant with cluster setting.
        # from_url method will send in the ip and port through a different variable then the
        # regular startup_nodes variable.
        if startup_nodes is None:
            if 'port' in connection_kwargs and 'host' in connection_kwargs:
                startup_nodes = [{
                    'host': connection_kwargs.pop('host'),
                    'port': str(connection_kwargs.pop('port')),
                }]

        self.max_connections = max_connections or 2 ** 31
        self.max_connections_per_node = max_connections_per_node
        self.nodes = NodeManager(
            startup_nodes,
            reinitialize_steps=reinitialize_steps,
            skip_full_coverage_check=skip_full_coverage_check,
            max_connections=self.max_connections,
            nodemanager_follow_cluster=nodemanager_follow_cluster,
            **connection_kwargs
        )
        self.initialized = False

        self.connections = {}
        self.connection_kwargs = connection_kwargs
        self.connection_kwargs['readonly'] = readonly
        self.readonly = readonly
        self.max_idle_time = max_idle_time
        self.idle_check_interval = idle_check_interval
        self.reset()

        if "stream_timeout" not in self.connection_kwargs:
            self.connection_kwargs["stream_timeout"] = ClusterConnectionPool.RedisClusterDefaultTimeout

    def __repr__(self):
        """
        Returns a string with all unique ip:port combinations that this pool
        is connected to
        """
        return "{0}<{1}>".format(
            type(self).__name__,
            ", ".join([self.connection_class.description.format(**node)
                       for node in self.nodes.startup_nodes])
        )

    async def initialize(self):
        if not self.initialized:
            await self.nodes.initialize()
            self.initialized = True

    async def disconnect_on_idle_time_exceeded(self, connection):
        while True:
            if (time.time() - connection.last_active_at > self.max_idle_time
                    and not connection.awaiting_response):
                connection.disconnect()
                node = connection.node
                self._available_connections[node['name']].remove(connection)
                self._created_connections_per_node[node['name']] -= 1
                break
            await asyncio.sleep(self.idle_check_interval)

    def reset(self):
        """Resets the connection pool back to a clean state"""
        self.pid = os.getpid()
        self._created_connections_per_node = {}  # Dict(Node, Int)
        self._available_connections = {}  # Dict(Node, List)
        self._in_use_connections = {}  # Dict(Node, Set)
        self._check_lock = threading.Lock()
        self.initialized = False

    def _checkpid(self):
        if self.pid != os.getpid():
            with self._check_lock:
                if self.pid == os.getpid():
                    # another thread already did the work while we waited
                    # on the lockself.
                    return
                self.disconnect()
                self.reset()

    def get_connection(self, command_name, *keys, **options):
        # Only pubsub command/connection should be allowed here
        if command_name != "pubsub":
            raise RedisClusterException("Only 'pubsub' commands can use get_connection()")

        channel = options.pop('channel', None)

        if not channel:
            return self.get_random_connection()

        slot = self.nodes.keyslot(channel)
        node = self.get_master_node_by_slot(slot)

        self._checkpid()

        try:
            connection = self._available_connections.get(node["name"], []).pop()
        except IndexError:
            connection = self.make_connection(node)

        if node['name'] not in self._in_use_connections:
            self._in_use_connections[node['name']] = set()

        self._in_use_connections[node['name']].add(connection)

        return connection

    def make_connection(self, node):
        """Creates a new connection"""
        if self.count_all_num_connections(node) >= self.max_connections:
            if self.max_connections_per_node:
                raise RedisClusterException("Too many connection ({0}) for node: {1}"
                                            .format(self.count_all_num_connections(node),
                                                    node['name']))

            raise RedisClusterException("Too many connections")

        self._created_connections_per_node.setdefault(node['name'], 0)
        self._created_connections_per_node[node['name']] += 1
        connection = self.connection_class(host=node["host"],
                                           port=node["port"],
                                           **self.connection_kwargs)

        # Must store node in the connection to make it eaiser to track
        connection.node = node
        if self.max_idle_time > self.idle_check_interval > 0:
            # do not await the future
            asyncio.ensure_future(self.disconnect_on_idle_time_exceeded(connection))
        return connection

    def release(self, connection):
        """Releases the connection back to the pool"""
        self._checkpid()
        if connection.pid != self.pid:
            return

        # Remove the current connection from _in_use_connection and add it back to the available pool
        # There is cases where the connection is to be removed but it will not exist and there
        # must be a safe way to remove
        i_c = self._in_use_connections.get(connection.node["name"], set())
        if connection in i_c:
            i_c.remove(connection)
        else:
            pass
        # discard connection with unread response
        if connection.awaiting_response:
            connection.disconnect()
            # reduce node connection count in case of too many connection error raised
            if self.max_connections_per_node and self._created_connections_per_node.get(connection.node['name']):
                self._created_connections_per_node[connection.node['name']] -= 1
        else:
            self._available_connections.setdefault(connection.node["name"], []).append(connection)

    def disconnect(self):
        """Closes all connectins in the pool"""
        all_conns = chain(
            self._available_connections.values(),
            self._in_use_connections.values(),
        )

        for node_connections in all_conns:
            for connection in node_connections:
                connection.disconnect()

    def count_all_num_connections(self, node):
        if self.max_connections_per_node:
            return self._created_connections_per_node.get(node['name'], 0)

        return sum([i for i in self._created_connections_per_node.values()])

    def get_random_connection(self):
        """Opens new connection to random redis server"""
        if self._available_connections:
            node_name = random.choice(list(self._available_connections.keys()))
            conn_list = self._available_connections[node_name]
            # check it in case of empty connection list
            if conn_list:
                return conn_list.pop()
        for node in self.nodes.random_startup_node_iter():
            connection = self.get_connection_by_node(node)

            if connection:
                return connection

        raise Exception("Cant reach a single startup node.")

    def get_connection_by_key(self, key):
        if not key:
            raise RedisClusterException("No way to dispatch this command to Redis Cluster.")

        return self.get_connection_by_slot(self.nodes.keyslot(key))

    def get_connection_by_slot(self, slot):
        """
        Determines what server a specific slot belongs to and return a redis
        object that is connected
        """
        self._checkpid()

        try:
            return self.get_connection_by_node(self.get_node_by_slot(slot))
        except KeyError:
            return self.get_random_connection()

    def get_connection_by_node(self, node):
        """Gets a connection by node"""
        self._checkpid()
        self.nodes.set_node_name(node)

        try:
            # Try to get connection from existing pool
            connection = self._available_connections.get(node["name"], []).pop()
        except IndexError:
            connection = self.make_connection(node)

        self._in_use_connections.setdefault(node["name"], set()).add(connection)

        return connection

    def get_master_node_by_slot(self, slot):
        return self.nodes.slots[slot][0]

    def get_node_by_slot(self, slot):
        if self.readonly:
            return random.choice(self.nodes.slots[slot])
        return self.get_master_node_by_slot(slot) class ConnectionPool:
    """Generic connection pool"""

    @classmethod
    def from_url(cls, url, db=None, decode_components=False, **kwargs):
        """
        Returns a connection pool configured from the given URL.

        For example::

        redis://[:password]@localhost:6379/0
        rediss://[:password]@localhost:6379/0
        unix://[:password]@/path/to/socket.sock?db=0

        Three URL schemes are supported:

        - ```redis://``
        <http://www.iana.org/assignments/uri-schemes/prov/redis>`_ creates a
        normal TCP socket connection
        - ```rediss://``
        <http://www.iana.org/assignments/uri-schemes/prov/rediss>`_ creates a
        SSL wrapped TCP socket connection
        - ``unix://`` creates a Unix Domain Socket connection

        There are several ways to specify a database number. The parse function
        will return the first specified option:
        1. A ``db`` querystring option, e.g. redis://localhost?db=0
        2. If using the redis:// scheme, the path argument of the url, e.g.
        redis://localhost/0
        3. The ``db`` argument to this function.

        If none of these options are specified, db=0 is used.

        The ``decode_components`` argument allows this function to work with
        percent-encoded URLs. If this argument is set to ``True`` all ``%xx``
        escapes will be replaced by their single-character equivalents after
        the URL has been parsed. This only applies to the ``hostname``,
        ``path``, and ``password`` components.

        Any additional querystring arguments and keyword arguments will be
        passed along to the ConnectionPool class's initializer. The querystring
        arguments ``connect_timeout`` and ``stream_timeout`` if supplied
        are parsed as float values. The arguments ``retry_on_timeout`` are
        parsed to boolean values that accept True/False, Yes/No values to indicate state.
        Invalid types cause a ``UserWarning`` to be raised.
        In the case of conflicting arguments, querystring arguments always win.
        """
        url = urlparse(url)
        qs = url.query

        url_options = {}

        for name, value in iter(parse_qs(qs).items()):
            if value and len(value) > 0:
                parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
                if parser:
                    try:
                        url_options[name] = parser(value[0])
                    except (TypeError, ValueError):
                        warnings.warn(UserWarning(
                            "Invalid value for `%s` in connection URL." % name
                        ))
                else:
                    url_options[name] = value[0]

        if decode_components:
            password = unquote(url.password) if url.password else None
            path = unquote(url.path) if url.path else None
            hostname = unquote(url.hostname) if url.hostname else None
        else:
            password = url.password
            path = url.path
            hostname = url.hostname

        # We only support redis:// and unix:// schemes.
        if url.scheme == 'unix':
            url_options.update({
                'password': password,
                'path': path,
                'connection_class': UnixDomainSocketConnection,
            })

        else:
            url_options.update({
                'host': hostname,
                'port': int(url.port or 6379),
                'password': password,
            })

            # If there's a path argument, use it as the db argument if a
            # querystring value wasn't specified
            if 'db' not in url_options and path:
                try:
                    url_options['db'] = int(path.replace('/', ''))
                except (AttributeError, ValueError):
                    pass

            if url.scheme == 'rediss':
                keyfile = url_options.pop('ssl_keyfile', None)
                certfile = url_options.pop('ssl_certfile', None)
                cert_reqs = url_options.pop('ssl_cert_reqs', None)
                ca_certs = url_options.pop('ssl_ca_certs', None)
                url_options['ssl_context'] = RedisSSLContext(keyfile, certfile, cert_reqs, ca_certs).get()

        # last shot at the db value
        url_options['db'] = int(url_options.get('db', db or 0))

        # update the arguments from the URL values
        kwargs.update(url_options)
        return cls(**kwargs)

    def __init__(self, connection_class=Connection, max_connections=None,
                 max_idle_time=0, idle_check_interval=1,
                 **connection_kwargs):
        """
        Creates a connection pool. If max_connections is set, then this
        object raises redis.ConnectionError when the pool's limit is reached.

        By default, TCP connections are created connection_class is specified.
        Use redis.UnixDomainSocketConnection for unix sockets.

        Any additional keyword arguments are passed to the constructor of
        connection_class.
        """
        max_connections = max_connections or 2 ** 31
        if not isinstance(max_connections, int) or max_connections < 0:
            raise ValueError('"max_connections" must be a positive integer')

        self.connection_class = connection_class
        self.connection_kwargs = connection_kwargs
        self.max_connections = max_connections
        self.max_idle_time = max_idle_time
        self.idle_check_interval = idle_check_interval
        self.loop = self.connection_kwargs.get('loop')

        self.reset()

    def __repr__(self):
        return '{}<{}>'.format(
            type(self).__name__,
            self.connection_class.description.format(**self.connection_kwargs),
        )

    async def disconnect_on_idle_time_exceeded(self, connection):
        while True:
            if (time.time() - connection.last_active_at > self.max_idle_time
                    and not connection.awaiting_response):
                connection.disconnect()
                try:
                    self._available_connections.remove(connection)
                except ValueError:
                    pass
                self._created_connections -= 1
                break
            await asyncio.sleep(self.idle_check_interval)

    def reset(self):
        self.pid = os.getpid()
        self._created_connections = 0
        self._available_connections = []
        self._in_use_connections = set()
        self._check_lock = threading.Lock()

    def _checkpid(self):
        if self.pid != os.getpid():
            with self._check_lock:
                if self.pid == os.getpid():
                    # another thread already did the work while we waited
                    # on the lock.
                    return
                self.disconnect()
                self.reset()

    def get_connection(self, *args, **kwargs):
        """Gets a connection from the pool"""
        self._checkpid()
        try:
            connection = self._available_connections.pop()
        except IndexError:
            connection = self.make_connection()
        self._in_use_connections.add(connection)
        return connection

    def make_connection(self):
        """Creates a new connection"""
        if self._created_connections >= self.max_connections:
            raise ConnectionError("Too many connections")
        self._created_connections += 1
        connection = self.connection_class(**self.connection_kwargs)
        if self.max_idle_time > self.idle_check_interval > 0:
            # do not await the future
            asyncio.ensure_future(self.disconnect_on_idle_time_exceeded(connection))
        return connection

    def release(self, connection):
        """Releases the connection back to the pool"""
        self._checkpid()
        if connection.pid != self.pid:
            return
        self._in_use_connections.remove(connection)
        # discard connection with unread response
        if connection.awaiting_response:
            connection.disconnect()
            self._created_connections -= 1
        else:
            self._available_connections.append(connection)

    def disconnect(self):
        """Closes all connections in the pool"""
        all_conns = chain(self._available_connections,
                          self._in_use_connections)
        for connection in all_conns:
            connection.disconnect()
            self._created_connections -= 1 class NodeFlag:
    BLOCKED = 'blocked'
    ALL_NODES = 'all-nodes'
    ALL_MASTERS = 'all-masters'
    RANDOM = 'random'
    SLOT_ID = 'slot-id' def blocked_command(self, command):
    """
    Raises a `RedisClusterException` mentioning the command is blocked.
    """
    raise RedisClusterException("Command: {0} is blocked in redis cluster mode".format(command)) def clusterdown_wrapper(func):
    """
    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 3 times to rerun the command and raises ClusterDownException if it continues to fail.
    """

    @wraps(func)
    async def inner(*args, **kwargs):
        for _ in range(0, 3):
            try:
                return await func(*args, **kwargs)
            except ClusterDownError:
                # Try again with the new cluster setup. All other errors
                # should be raised.
                pass

        # If it fails 3 times then raise exception back to caller
        raise ClusterDownError("CLUSTERDOWN error. Unable to rebuild the cluster")

    return inner def dict_merge(*dicts):
    merged = {}
    for d in dicts:
        merged.update(d)
    return merged def first_key(res):
    """
    Returns the first result for the given command.

    If more then 1 result is returned then a `RedisClusterException` is raised.
    """
    if not isinstance(res, dict):
        raise ValueError('Value should be of dict type')

    if len(res.keys()) != 1:
        raise RedisClusterException("More then 1 result from command")

    return list(res.values())[0]
import asyncio
import sys
from aredis.commands.cluster import ClusterCommandMixin
from aredis.commands.connection import ClusterConnectionCommandMixin, ConnectionCommandMixin
from aredis.commands.extra import ExtraCommandMixin
from aredis.commands.geo import GeoCommandMixin
from aredis.commands.hash import ClusterHashCommandMixin, HashCommandMixin
from aredis.commands.hyperlog import ClusterHyperLogCommandMixin, HyperLogCommandMixin
from aredis.commands.keys import ClusterKeysCommandMixin, KeysCommandMixin
from aredis.commands.lists import ClusterListsCommandMixin, ListsCommandMixin
from aredis.commands.pubsub import CLusterPubSubCommandMixin, PubSubCommandMixin
from aredis.commands.scripting import ClusterScriptingCommandMixin, ScriptingCommandMixin
from aredis.commands.sentinel import ClusterSentinelCommands, SentinelCommandMixin
from aredis.commands.server import ClusterServerCommandMixin, ServerCommandMixin
from aredis.commands.sets import ClusterSetsCommandMixin, SetsCommandMixin
from aredis.commands.sorted_set import ClusterSortedSetCommandMixin, SortedSetCommandMixin
from aredis.commands.streams import StreamsCommandMixin
from aredis.commands.strings import ClusterStringsCommandMixin, StringsCommandMixin
from aredis.commands.transaction import ClusterTransactionCommandMixin, TransactionCommandMixin
from aredis.compat import CancelledError
from aredis.connection import RedisSSLContext, UnixDomainSocketConnection
from aredis.exceptions import (AskError, BusyLoadingError, ClusterDownError, ClusterError, ConnectionError, MovedError,
                               RedisClusterException, TimeoutError, TryAgainError)
from aredis.pool import (ClusterConnectionPool, ConnectionPool)
from aredis.utils import (NodeFlag, blocked_command, clusterdown_wrapper, dict_merge, first_key)
    from aredis.commands.iter import IterCommandMixin, ClusterIterCommandMixin
        from aredis.pipeline import StrictPipeline
        from aredis.pipeline import StrictClusterPipeline
        "Sets a custom Result Callback"
        self.result_callbacks[command] = callback

    def _determine_slot(self, *args):
        """Figures out what slot based on command and args"""
        if len(args) <= 1:
            raise RedisClusterException("No way to dispatch this command to Redis Cluster. Missing key.")
        command = args[0]

        if command in ['EVAL', 'EVALSHA']:
            numkeys = args[2]
            keys = args[3: 3 + numkeys]
            slots = {self.connection_pool.nodes.keyslot(key) for key in keys}
            if len(slots) != 1:
                raise RedisClusterException("{0} - all keys must map to the same key slot".format(command))
            return slots.pop()
        elif command in ('XREAD', 'XREADGROUP'):
            try:
                idx = args.index('STREAMS') + 1
            except ValueError:
                raise RedisClusterException("{0} arguments do not contain STREAMS operand".format(command))
            key = args[idx]
        elif command in ('XGROUP', 'XINFO'):
            key = args[2]
        else:
            key = args[1]

        return self.connection_pool.nodes.keyslot(key)

    def _merge_result(self, command, res, **kwargs):
        """
        `res` is a dict with the following structure Dict(NodeName, CommandResult)
        """
        if command in self.result_callbacks:
            return self.result_callbacks[command](res, **kwargs)

        # Default way to handle result
        return first_key(res)

    def determine_node(self, *args, **kwargs):
        """
        TODO: document
        """
        command = args[0]
        node_flag = self.nodes_flags.get(command)

        if node_flag == NodeFlag.BLOCKED:
            return blocked_command(self, command)
        elif node_flag == NodeFlag.RANDOM:
            return [self.connection_pool.nodes.random_node()]
        elif node_flag == NodeFlag.ALL_MASTERS:
            return self.connection_pool.nodes.all_masters()
        elif node_flag == NodeFlag.ALL_NODES:
            return self.connection_pool.nodes.all_nodes()
        elif node_flag == NodeFlag.SLOT_ID:
            # if node flag of command is SLOT_ID
            # `slot_id` should is assumed in kwargs
            slot = kwargs.get('slot_id')
            if not slot:
                raise RedisClusterException('slot_id is needed to execute command {}'
                                            .format(command))
            return [self.connection_pool.nodes.node_from_slot(slot)]
        else:
            return None

    @clusterdown_wrapper
    async def execute_command(self, *args, **kwargs):
        """
        Sends a command to a node in the cluster
        """
        if not self.connection_pool.initialized:
            await self.connection_pool.initialize()
        if not args:
            raise RedisClusterException("Unable to determine command to use")

        command = args[0]

        node = self.determine_node(*args, **kwargs)
        if node:
            return await self.execute_command_on_nodes(node, *args, **kwargs)

        # If set externally we must update it before calling any commands
        if self.refresh_table_asap:
            await self.connection_pool.nodes.initialize()
            self.refresh_table_asap = False

        redirect_addr = None
        asking = False

        try_random_node = False
        slot = self._determine_slot(*args)
        ttl = int(self.RedisClusterRequestTTL)

        while ttl > 0:
            ttl -= 1

            if asking:
                node = self.connection_pool.nodes.nodes[redirect_addr]
                r = self.connection_pool.get_connection_by_node(node)
            elif try_random_node:
                r = self.connection_pool.get_random_connection()
                try_random_node = False
            else:
                if self.refresh_table_asap:
                    # MOVED
                    node = self.connection_pool.get_master_node_by_slot(slot)
                else:
                    node = self.connection_pool.get_node_by_slot(slot)
                r = self.connection_pool.get_connection_by_node(node)

            try:
                if asking:
                    await r.send_command('ASKING')
                    await self.parse_response(r, "ASKING", **kwargs)
                    asking = False

                await r.send_command(*args)
                return await self.parse_response(r, command, **kwargs)
            except (RedisClusterException, BusyLoadingError, CancelledError):
                raise

            except (ConnectionError, TimeoutError):