k3redisutil

Redis utilities for easier client management and proxy support.

k3redisutil is a component of pykit3 project: a python3 toolkit set.

Installation

pip install k3redisutil

Quick Start

import k3redisutil

# Get a process-wise singleton redis client
client = k3redisutil.get_client(6379)
client.set('foo', 'bar')

# Using redis as a duplex cross-process channel
c = k3redisutil.RedisChannel(6379, '/foo', 'client')
s = k3redisutil.RedisChannel(6379, '/foo', 'server')

c.send_msg('hello from client')
print(s.recv_msg())  # 'hello from client'

# Using redis proxy client
cli = k3redisutil.RedisProxyClient([('127.0.0.1', 2222)])
cli.set('key', 'value', expire=1000)  # with TTL in msec
print(cli.get('key'))

API Reference

k3redisutil

For using redis more easily.

KeyNotFoundError

Bases: RedisProxyError

It is a subclass of redisutil.RedisProxyError. Raise if key not found (redis proxy server return a 404).

Source code in k3redisutil/redis_proxy_cli.py

class KeyNotFoundError(RedisProxyError):
    """
    It is a subclass of `redisutil.RedisProxyError`.
    Raise if key not found (redis proxy server return a `404`).
    """

    pass

RedisChannel

Bases: object

send message data through channel. channel is a list in redis. peer is "client" or "server".

send is a rpush operation that adds an item to the end of a list. recv is a lpop operation that pops an item from the start of a list.

Source code in k3redisutil/redisutil.py

class RedisChannel(object):
    """
    send message `data` through `channel`.
    `channel` is a list in redis.
    `peer` is "client" or "server".

    send is a rpush operation that adds an item to the end of a list.
    recv is a lpop operation that pops an item from the start of a list.
    """

    other_peer = {
        "client": "server",
        "server": "client",
    }

    def __init__(self, ip_port, channel, peer, timeout=None):
        """
        `redisutil.RedisChannel(ip_port, channel, peer, timeout=None)`

        Create a redis list based channel for cross process communication.

        See [redis-list-command](https://redis.io/commands#list).

        Initializing this class does **NOT** create a socket connecting to redis or
        create any data in redis.

        A channel support duplex communication.
        Thus in redis it creates two lists for each channel for two way communication:
        `<channel>/client` and `<channel>/server`.

        Client side uses `<channel>/client` to send message.
        Server side uses `<channel>/server` to send message.

        A client should initialize `RedisChannel` with argument `peer` set to `client`.
        A server should initialize `RedisChannel` with argument `peer` set to `server`.
        :param ip_port: tuple of `(ip, port)` or a single int/long number as port.
        :param channel: specifies the name of the channel to create.
        A channel should be in URI form, starting with `/`, such as: `/asyncworker/image-process`
        `channel` can also be a tuple of string:
        `('asyncworker', 'image-process')` will be converted to `/asyncworker/image-process`.
        :param peer: specifies this instance is a client or a server.
        It can be `"client"` or `"server"`.
        :param timeout: the expire time of the channel, in second.
        If it is `None`, the channel exists until being cleaned.
        """
        assert peer in self.other_peer

        self.ip_port = normalize_ip_port(ip_port)
        self.rcl = get_client(self.ip_port)

        # convert ['a', 'b'] to '/a/b'
        if isinstance(channel, (list, tuple)):
            channel = "/" + "/".join(channel)

        self.channel = channel
        self.peer = peer.lower()
        self.send_list_name = "/".join([self.channel, self.peer])
        self.recv_list_name = "/".join([self.channel, self.other_peer[self.peer]])
        self.timeout = timeout

    def send_msg(self, data):
        """
        Send data. `data` is json-encoded in redis list.
        :param data: any data type that can be json encoded.
        :return: Nothing
        """
        j = k3utfjson.dump(data)
        self.rcl.rpush(self.send_list_name, j)

        if self.timeout is not None:
            self.rcl.expire(self.send_list_name, self.timeout)

    def recv_msg(self):
        """
        Receive one message.
        If there is no message in this channel, it returns `None`.
        :return: data that is loaded from json. Or `None` if there is no message in channel.
        """
        v = self.rcl.lpop(self.recv_list_name)
        if v is None:
            return None

        return k3utfjson.load(v)

    def brecv_msg(self, timeout=0):
        """
        Block to receive one message.
        If there is no message in this channel,
        then block for `timeout` seconds or until
        a value was pushed to the channel.

        :param timeout: seconds of the block time.
        If it is `0`, then block until a message is available.

        :return: data that is loaded from json. Or `None` if timeout.
        """
        v = self.rcl.blpop(self.recv_list_name, timeout=timeout)
        if v is None or len(v) != 2:
            return None

        # v is a tuple, (key, value)
        _, value = v

        return k3utfjson.load(value)

    def recv_last_msg(self):
        """
        Similar to `RedisChannel.recv_msg` except it returns only the last message it
        sees and removes all previous messages from this channel.
        :return: data that is loaded from json. Or `None` if there is no message in channel.
        """
        last = None
        while True:
            v = self.rcl.lpop(self.recv_list_name)
            if v is None:
                return k3utfjson.load(last)

            last = v

    def brecv_last_msg(self, timeout=0):
        """
        Similar to `RedisChannel.recv_last_msg` except it blocks for `timeout`
        seconds if the channel is empty.
        :param timeout: seconds of the block time.
        If it is `0`, then block until a message is available.
        :return: data that is loaded from json. Or `None` if timeout.
        """
        msg = self.brecv_msg(timeout=timeout)
        if msg is None:
            return None

        last = self.recv_last_msg() or msg

        return last

    def peek_msg(self):
        """
        Similar to `RedisChannel.recv_msg` except it does not remove the message it
        returned.
        :return: data that is loaded from json. Or `None` if there is no message in channel.
        """
        v = self.rcl.lindex(self.recv_list_name, 0)
        if v is None:
            return None

        return k3utfjson.load(v)

    def rpeek_msg(self):
        """
        Similar to `RedisChannel.peek_msg`
        except it gets message from the tail of the channel.
        :return: JSON decoded message. Or `None` if there is no message in channel.
        """
        v = self.rcl.lindex(self.recv_list_name, -1)
        if v is None:
            return None

        return k3utfjson.load(v)

    def list_channel(self, prefix):
        """
        List all channel names those start with `prefix`.
        :param prefix: specifies what channel to list. `prefix` must starts with '/', ends with '/'.
        :return: a list of channel names.
        """
        if isinstance(prefix, (list, tuple)):
            _prefix = "/" + "/".join(prefix) + "/"
        else:
            _prefix = prefix

        if not _prefix.startswith("/"):
            raise ValueError('prefix must starts with "/", but:' + repr(prefix))

        if _prefix.endswith("*"):
            raise ValueError('prefix must NOT ends with "*", but:' + repr(prefix))

        if not _prefix.endswith("/"):
            raise ValueError('prefix must ends with "/", but:' + repr(prefix))

        _prefix = _prefix + "*"
        channels = self.rcl.keys(_prefix)

        rst = []
        for c in channels:
            for k in self.other_peer:
                k = "/" + k
                if c.endswith(k.encode()):
                    c = c[: -len(k)]
                    break
            else:
                logger.info("not a channel: " + repr(c))
                continue

            if c not in rst:
                rst.append(c)

        return sorted(rst)

__init__(ip_port, channel, peer, timeout=None)

redisutil.RedisChannel(ip_port, channel, peer, timeout=None)

Create a redis list based channel for cross process communication.

See redis-list-command.

Initializing this class does NOT create a socket connecting to redis or create any data in redis.

A channel support duplex communication. Thus in redis it creates two lists for each channel for two way communication: <channel>/client and <channel>/server.

Client side uses <channel>/client to send message. Server side uses <channel>/server to send message.

A client should initialize RedisChannel with argument peer set to client. A server should initialize RedisChannel with argument peer set to server. :param ip_port: tuple of (ip, port) or a single int/long number as port. :param channel: specifies the name of the channel to create. A channel should be in URI form, starting with /, such as: /asyncworker/image-process channel can also be a tuple of string: ('asyncworker', 'image-process') will be converted to /asyncworker/image-process. :param peer: specifies this instance is a client or a server. It can be "client" or "server". :param timeout: the expire time of the channel, in second. If it is None, the channel exists until being cleaned.

Source code in k3redisutil/redisutil.py

def __init__(self, ip_port, channel, peer, timeout=None):
    """
    `redisutil.RedisChannel(ip_port, channel, peer, timeout=None)`

    Create a redis list based channel for cross process communication.

    See [redis-list-command](https://redis.io/commands#list).

    Initializing this class does **NOT** create a socket connecting to redis or
    create any data in redis.

    A channel support duplex communication.
    Thus in redis it creates two lists for each channel for two way communication:
    `<channel>/client` and `<channel>/server`.

    Client side uses `<channel>/client` to send message.
    Server side uses `<channel>/server` to send message.

    A client should initialize `RedisChannel` with argument `peer` set to `client`.
    A server should initialize `RedisChannel` with argument `peer` set to `server`.
    :param ip_port: tuple of `(ip, port)` or a single int/long number as port.
    :param channel: specifies the name of the channel to create.
    A channel should be in URI form, starting with `/`, such as: `/asyncworker/image-process`
    `channel` can also be a tuple of string:
    `('asyncworker', 'image-process')` will be converted to `/asyncworker/image-process`.
    :param peer: specifies this instance is a client or a server.
    It can be `"client"` or `"server"`.
    :param timeout: the expire time of the channel, in second.
    If it is `None`, the channel exists until being cleaned.
    """
    assert peer in self.other_peer

    self.ip_port = normalize_ip_port(ip_port)
    self.rcl = get_client(self.ip_port)

    # convert ['a', 'b'] to '/a/b'
    if isinstance(channel, (list, tuple)):
        channel = "/" + "/".join(channel)

    self.channel = channel
    self.peer = peer.lower()
    self.send_list_name = "/".join([self.channel, self.peer])
    self.recv_list_name = "/".join([self.channel, self.other_peer[self.peer]])
    self.timeout = timeout

brecv_last_msg(timeout=0)

Similar to RedisChannel.recv_last_msg except it blocks for timeout seconds if the channel is empty. :param timeout: seconds of the block time. If it is 0, then block until a message is available. :return: data that is loaded from json. Or None if timeout.

Source code in k3redisutil/redisutil.py

def brecv_last_msg(self, timeout=0):
    """
    Similar to `RedisChannel.recv_last_msg` except it blocks for `timeout`
    seconds if the channel is empty.
    :param timeout: seconds of the block time.
    If it is `0`, then block until a message is available.
    :return: data that is loaded from json. Or `None` if timeout.
    """
    msg = self.brecv_msg(timeout=timeout)
    if msg is None:
        return None

    last = self.recv_last_msg() or msg

    return last

brecv_msg(timeout=0)

Block to receive one message. If there is no message in this channel, then block for timeout seconds or until a value was pushed to the channel.

:param timeout: seconds of the block time. If it is 0, then block until a message is available.

:return: data that is loaded from json. Or None if timeout.

Source code in k3redisutil/redisutil.py

def brecv_msg(self, timeout=0):
    """
    Block to receive one message.
    If there is no message in this channel,
    then block for `timeout` seconds or until
    a value was pushed to the channel.

    :param timeout: seconds of the block time.
    If it is `0`, then block until a message is available.

    :return: data that is loaded from json. Or `None` if timeout.
    """
    v = self.rcl.blpop(self.recv_list_name, timeout=timeout)
    if v is None or len(v) != 2:
        return None

    # v is a tuple, (key, value)
    _, value = v

    return k3utfjson.load(value)

list_channel(prefix)

List all channel names those start with prefix. :param prefix: specifies what channel to list. prefix must starts with '/', ends with '/'. :return: a list of channel names.

Source code in k3redisutil/redisutil.py

def list_channel(self, prefix):
    """
    List all channel names those start with `prefix`.
    :param prefix: specifies what channel to list. `prefix` must starts with '/', ends with '/'.
    :return: a list of channel names.
    """
    if isinstance(prefix, (list, tuple)):
        _prefix = "/" + "/".join(prefix) + "/"
    else:
        _prefix = prefix

    if not _prefix.startswith("/"):
        raise ValueError('prefix must starts with "/", but:' + repr(prefix))

    if _prefix.endswith("*"):
        raise ValueError('prefix must NOT ends with "*", but:' + repr(prefix))

    if not _prefix.endswith("/"):
        raise ValueError('prefix must ends with "/", but:' + repr(prefix))

    _prefix = _prefix + "*"
    channels = self.rcl.keys(_prefix)

    rst = []
    for c in channels:
        for k in self.other_peer:
            k = "/" + k
            if c.endswith(k.encode()):
                c = c[: -len(k)]
                break
        else:
            logger.info("not a channel: " + repr(c))
            continue

        if c not in rst:
            rst.append(c)

    return sorted(rst)

peek_msg()

Similar to RedisChannel.recv_msg except it does not remove the message it returned. :return: data that is loaded from json. Or None if there is no message in channel.

Source code in k3redisutil/redisutil.py

def peek_msg(self):
    """
    Similar to `RedisChannel.recv_msg` except it does not remove the message it
    returned.
    :return: data that is loaded from json. Or `None` if there is no message in channel.
    """
    v = self.rcl.lindex(self.recv_list_name, 0)
    if v is None:
        return None

    return k3utfjson.load(v)

recv_last_msg()

Similar to RedisChannel.recv_msg except it returns only the last message it sees and removes all previous messages from this channel. :return: data that is loaded from json. Or None if there is no message in channel.

Source code in k3redisutil/redisutil.py

def recv_last_msg(self):
    """
    Similar to `RedisChannel.recv_msg` except it returns only the last message it
    sees and removes all previous messages from this channel.
    :return: data that is loaded from json. Or `None` if there is no message in channel.
    """
    last = None
    while True:
        v = self.rcl.lpop(self.recv_list_name)
        if v is None:
            return k3utfjson.load(last)

        last = v

recv_msg()

Receive one message. If there is no message in this channel, it returns None. :return: data that is loaded from json. Or None if there is no message in channel.

Source code in k3redisutil/redisutil.py

def recv_msg(self):
    """
    Receive one message.
    If there is no message in this channel, it returns `None`.
    :return: data that is loaded from json. Or `None` if there is no message in channel.
    """
    v = self.rcl.lpop(self.recv_list_name)
    if v is None:
        return None

    return k3utfjson.load(v)

rpeek_msg()

Similar to RedisChannel.peek_msg except it gets message from the tail of the channel. :return: JSON decoded message. Or None if there is no message in channel.

Source code in k3redisutil/redisutil.py

def rpeek_msg(self):
    """
    Similar to `RedisChannel.peek_msg`
    except it gets message from the tail of the channel.
    :return: JSON decoded message. Or `None` if there is no message in channel.
    """
    v = self.rcl.lindex(self.recv_list_name, -1)
    if v is None:
        return None

    return k3utfjson.load(v)

send_msg(data)

Send data. data is json-encoded in redis list. :param data: any data type that can be json encoded. :return: Nothing

Source code in k3redisutil/redisutil.py

def send_msg(self, data):
    """
    Send data. `data` is json-encoded in redis list.
    :param data: any data type that can be json encoded.
    :return: Nothing
    """
    j = k3utfjson.dump(data)
    self.rcl.rpush(self.send_list_name, j)

    if self.timeout is not None:
        self.rcl.expire(self.send_list_name, self.timeout)

RedisProxyClient

Bases: object

redis operation, http method, count of args, optional args name

### RedisProxyClient.delete Delete the specified key. Do nothing if the key doesn’t exist.

arguments: key:specifies the key to redis. retry:try to send request for another N times while failed to send request. By default, it is 0.

return: None ### RedisProxyClient.get Get the value with key. Raise a redisutil.KeyNotFoundError if the key doesn’t exist.

retry:try to send request for another N times while failed to send request. By default, it is 0.

return: the value at key.

### RedisProxyClient.hdel

Delete the specified hashkey in the specified hashname. Raise a redisutil.KeyNotFoundError if it doesn’t exist.

arguments: hashname:specifies the hash name to redis.

hashkey:specifies the hash key to redis.

retry:try to send request for another N times while failed to send request. By default, it is 0.

return: None

### RedisProxyClient.set

Set the value at key to val. arguments: key: specifies the key to redis. val: specifies the value to redis. expire: the expire time on key in msec. Defaults to None. retry: try to send request for another N times while failed to send request. By default, it is 0.

return: nothing

### RedisProxyClient.hget

Return the value of haskkey within the haskname. Raise a redisutil.KeyNotFoundError if it doesn’t exist.

arguments: hashname: specifies the hash name to redis. hashkey: specifies the hash key to redis. retry: try to send request for another N times while failed to send request. By default, it is 0.

return: the value of hashkey within the hashname.

### RedisProxyClient.hset

Set hashkey to val within hashname.

arguments:

hashname: specifies the hash name to redis.

hashkey: specifies the hash key to redis.

val: specifies the value to redis.

expire: the expire time on hashkey within hashname in msec. Defaults to None.

retry: try to send request for another N times while failed to send request. By default, it is 0.

return: nothing

### RedisProxyClient.hkeys

Return the list of keys within hashname. Raise a redisutil.KeyNotFoundError if the hashname doesn’t exist.

arguments:

hashname: specifies the hash name to redis.

retry: try to send request for another N times while failed to send request. By default, it is 0.

return: a list contains all the keys.

### RedisProxyClient.hvals

Return the list of values within hashname. Raise a redisutil.KeyNotFoundError if the hashname doesn’t exist.

arguments:

hashname: specifies the hash name to redis.

retry: try to send request for another N times while failed to send request. By default, it is 0.

return: a list contains all the values.

### RedisProxyClient.hgetall

Return a dict of the hash’s name/value pairs. Raise a redisutil.KeyNotFoundError if the hashname doesn’t exist.

arguments:

hashname:specifies the hash name to redis.

retry:try to send request for another N times while failed to send request. By default, it is 0.

return: a dict of the hash’s name/value pairs.

Source code in k3redisutil/redis_proxy_cli.py

class RedisProxyClient(object):
    """
     redis operation, http method, count of args, optional args name

     ### RedisProxyClient.delete
     Delete the specified key.
     Do nothing if the key doesn’t exist.

     **arguments**:
     `key`:specifies the key to redis.
     `retry`:try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: None
     ###  RedisProxyClient.get
     Get the value with `key`. Raise a `redisutil.KeyNotFoundError` if the key doesn’t exist.

    `retry`:try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: the value at `key`.

     ### RedisProxyClient.hdel

     Delete the specified `hashkey` in the specified `hashname`.
     Raise a `redisutil.KeyNotFoundError` if it doesn’t exist.

     **arguments**:
     `hashname`:specifies the hash name to redis.

     `hashkey`:specifies the hash key to redis.

     `retry`:try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: None

     ###  RedisProxyClient.set

     Set the value at `key` to `val`.
     **arguments**:
     `key`: specifies the key to redis.
     `val`: specifies the value to redis.
     `expire`: the expire time on `key` in msec. Defaults to `None`.
     `retry`: try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: nothing

     ###  RedisProxyClient.hget

     Return the value of `haskkey` within the `haskname`.
     Raise a `redisutil.KeyNotFoundError` if it doesn’t exist.

     **arguments**:
     `hashname`: specifies the hash name to redis.
     `hashkey`: specifies the hash key to redis.
     `retry`: try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: the value of `hashkey` within the `hashname`.

     ###  RedisProxyClient.hset

     Set `hashkey` to `val` within `hashname`.

     **arguments**:

     `hashname`: specifies the hash name to redis.

     `hashkey`: specifies the hash key to redis.

     `val`: specifies the value to redis.

     `expire`: the expire time on `hashkey` within `hashname` in msec. Defaults to `None`.

     `retry`: try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: nothing

     ###  RedisProxyClient.hkeys

     Return the list of keys within `hashname`.
     Raise a `redisutil.KeyNotFoundError` if the `hashname` doesn’t exist.

     **arguments**:

     `hashname`: specifies the hash name to redis.

     `retry`: try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: a `list` contains all the keys.

     ###  RedisProxyClient.hvals

     Return the list of values within `hashname`.
     Raise a `redisutil.KeyNotFoundError` if the `hashname` doesn’t exist.

     **arguments**:

     `hashname`: specifies the hash name to redis.

     `retry`: try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: a `list` contains all the values.

     ###  RedisProxyClient.hgetall

     Return a dict of the hash’s name/value pairs.
     Raise a `redisutil.KeyNotFoundError` if the `hashname` doesn’t exist.

     **arguments**:

     `hashname`:specifies the hash name to redis.

     `retry`:try to send request for another N times while failed to send request.
     By default, it is `0`.

     **return**: a `dict` of the hash’s name/value pairs.
    """

    methods = {
        # get(key, retry=0)
        "get": ("get", "GET", 2, ()),
        # set(key, val, expire=None, retry=0)
        "set": ("set", "PUT", 4, ("expire",)),
        # hget(hashname, hashkey, retry=0)
        "hget": ("hget", "GET", 3, ()),
        # hset(hashname, hashkey, val, expire=None, retry=0)
        "hset": ("hset", "PUT", 5, ("expire",)),
        # hkeys(hashname, retry=0)
        "hkeys": ("hkeys", "GET", 2, ()),
        # hvals(hashname, retry=0)
        "hvals": ("hvals", "GET", 2, ()),
        # hgetall(hashname, retry=0)
        "hgetall": ("hgetall", "GET", 2, ()),
        # delete(key, retry=0)
        "delete": ("del", "DELETE", 2, ()),
        # hdel(hashname, key, retry=0)
        "hdel": ("hdel", "DELETE", 3, ()),
    }

    def __init__(self, hosts, proxy_hosts=None, nwr=None, ak_sk=None, timeout=None):
        """
        A client for redis proxy server.
        :param hosts: a `tuple` each element is a `(ip, port)`, retry with next addr while failed to use prev one.
        :param proxy_hosts: a `tuple` each element is `hosts`, as the backup hosts.
        :param nwr: `tuple` of `(n, w, r)`, count of replicas, count of write, count of read.
        If it is `None`, `config.rp_cli_nwr` is used.
        :param ak_sk: tuple` of `(access_key, secret_key)` to sign the request.
        If it is `None`, `config.rp_cli_ak_sk` is used.
        :param timeout: timeout of connecting redis proxy server in seconds. Defaults to `None`.
        """
        self.hosts = hosts
        self.proxy_hosts = proxy_hosts

        if nwr is None:
            nwr = conf.rp_cli_nwr

        if ak_sk is None:
            ak_sk = conf.rp_cli_ak_sk

        self.n, self.w, self.r = nwr
        self.access_key, self.secret_key = ak_sk

        self.timeout = timeout or DEFAULT_TIMEOUT
        self.ver = "/redisproxy/v1"

        for mtd_name, mtd_info in self.methods.items():
            api_obj = SetAPI(self, mtd_info[0], mtd_info[1:])
            setattr(self, mtd_name, api_obj.api)

    def _sign_req(self, req):
        sign_payload = True if "body" in req else False
        signer = k3awssign.Signer(self.access_key, self.secret_key)
        sign_ctx = signer.add_auth(req, query_auth=True, sign_payload=sign_payload)
        logger.debug("signing details: {ctx}".format(ctx=sign_ctx))

    def _make_req_uri(self, params, qs):
        path = [self.ver]
        path.extend(params)

        qs_list = [
            "n={n}".format(n=self.n),
            "w={w}".format(w=self.w),
            "r={r}".format(r=self.r),
        ]
        for k, v in qs.items():
            if v is None:
                continue

            qs_list.append("{k}={v}".format(k=k, v=v))

        return "{p}?{qs}".format(p="/".join(path), qs="&".join(qs_list))

    def _req(self, req):
        if "headers" not in req:
            req["headers"] = {
                "host": "{ip}:{port}".format(ip=self.ip, port=self.port),
            }

        elif "host" not in req["headers"]:
            req["headers"]["host"] = "{ip}:{port}".format(ip=self.ip, port=self.port)

        body = req.get("body", None)
        if body is not None:
            req["headers"]["Content-Length"] = len(body)

        self._sign_req(req)

        cli = k3http.Client(self.ip, self.port, self.timeout)
        cli.send_request(req["uri"], method=req["verb"], headers=req["headers"])
        if body is not None:
            cli.send_body(req["body"])

        cli.read_response()
        res = cli.read_body(None)

        msg = "Status:{s} req:{req} res:{res} server:{ip}:{p}".format(
            s=cli.status, req=repr(req), res=repr(res), ip=self.ip, p=self.port
        )

        if cli.status == 404:
            raise KeyNotFoundError(msg)

        elif cli.status != 200:
            raise ServerResponseError(msg)

        return res

    @_proxy
    @_retry
    def _api(self, verb, path, body, qs):
        req = {
            "verb": verb,
            "uri": self._make_req_uri(path, qs),
        }

        if body is not None:
            req["body"] = body

        if verb == "GET":
            rst = self._req(req)
            return k3utfjson.load(rst)

        else:
            self._req(req)

__init__(hosts, proxy_hosts=None, nwr=None, ak_sk=None, timeout=None)

A client for redis proxy server. :param hosts: a tuple each element is a (ip, port), retry with next addr while failed to use prev one. :param proxy_hosts: a tuple each element is hosts, as the backup hosts. :param nwr: tuple of (n, w, r), count of replicas, count of write, count of read. If it is None, config.rp_cli_nwr is used. :param ak_sk: tupleof(access_key, secret_key)to sign the request. If it isNone,config.rp_cli_ak_skis used. :param timeout: timeout of connecting redis proxy server in seconds. Defaults toNone`.

Source code in k3redisutil/redis_proxy_cli.py

def __init__(self, hosts, proxy_hosts=None, nwr=None, ak_sk=None, timeout=None):
    """
    A client for redis proxy server.
    :param hosts: a `tuple` each element is a `(ip, port)`, retry with next addr while failed to use prev one.
    :param proxy_hosts: a `tuple` each element is `hosts`, as the backup hosts.
    :param nwr: `tuple` of `(n, w, r)`, count of replicas, count of write, count of read.
    If it is `None`, `config.rp_cli_nwr` is used.
    :param ak_sk: tuple` of `(access_key, secret_key)` to sign the request.
    If it is `None`, `config.rp_cli_ak_sk` is used.
    :param timeout: timeout of connecting redis proxy server in seconds. Defaults to `None`.
    """
    self.hosts = hosts
    self.proxy_hosts = proxy_hosts

    if nwr is None:
        nwr = conf.rp_cli_nwr

    if ak_sk is None:
        ak_sk = conf.rp_cli_ak_sk

    self.n, self.w, self.r = nwr
    self.access_key, self.secret_key = ak_sk

    self.timeout = timeout or DEFAULT_TIMEOUT
    self.ver = "/redisproxy/v1"

    for mtd_name, mtd_info in self.methods.items():
        api_obj = SetAPI(self, mtd_info[0], mtd_info[1:])
        setattr(self, mtd_name, api_obj.api)

RedisProxyError

Bases: Exception

The base class of other exceptions of RedisProxyClient. It is a subclass of Exception.

Source code in k3redisutil/redis_proxy_cli.py

class RedisProxyError(Exception):
    """
    The base class of other exceptions of `RedisProxyClient`.
    It is a subclass of `Exception`.
    """

    pass

SendRequestError

Bases: RedisProxyError

It is a subclass of redisutil.RedisProxyError. Raise if failed to send request to redis proxy server.

Source code in k3redisutil/redis_proxy_cli.py

class SendRequestError(RedisProxyError):
    """
    It is a subclass of `redisutil.RedisProxyError`.
    Raise if failed to send request to redis proxy server.
    """

    pass

ServerResponseError

Bases: RedisProxyError

It is a subclass of redisutil.RedisProxyError. Raise if http-status not in (200, 404) return from redis proxy server.

Source code in k3redisutil/redis_proxy_cli.py

class ServerResponseError(RedisProxyError):
    """
    It is a subclass of `redisutil.RedisProxyError`.
    Raise if http-status not in `(200, 404)` return from redis proxy server.

    """

    pass

get_client(ip_port)

Return a process-wise singleton redis client, which is an instance of redis.StrictRedis.

Redis client returned is shared across the entire process and will not be re-created. It is also safe to use if a process fork and inherited opened socket file-descriptors. :param ip_port: could be a port number in int or long to get or create a client connecting to localhost. It can also be tuple of (ip, port). :return: an instance of redis.StrictRedis.

Source code in k3redisutil/redisutil.py

def get_client(ip_port):
    """
    Return a process-wise singleton redis client, which is an instance of `redis.StrictRedis`.

    Redis client returned is shared across the entire process and will not be
    re-created.
    It is also safe to use if a process fork and inherited opened socket file-descriptors.
    :param ip_port: could be a port number in `int` or `long` to get or create a
    client connecting to localhost.
    It can also be tuple of `(ip, port)`.
    :return: an instance of `redis.StrictRedis`.
    """
    ip_port = normalize_ip_port(ip_port)

    pid = os.getpid()

    with _lock:
        o = _pid_client[ip_port]

        if pid not in o:
            o[pid] = redis.StrictRedis(*ip_port)

    return _pid_client[ip_port][pid]

wait_serve(ip_port, timeout=5)

Wait for at most timeout seconds until redis start serving request. It is useful when start a redis server. :param ip_port: tuple of (ip, port) or a single int/long number as port. :param timeout: specifies max waiting time, in seconds. :return: Nothing

Source code in k3redisutil/redisutil.py

def wait_serve(ip_port, timeout=5):
    """
    Wait for at most `timeout` seconds until redis start serving request.
    It is useful when start a redis server.
    :param ip_port: tuple of `(ip, port)` or a single int/long number as port.
    :param timeout: specifies max waiting time, in seconds.
    :return: Nothing
    """
    t = time.time() + timeout

    rcl = get_client(ip_port)

    while time.time() < t:
        try:
            rcl.hget("foo", "foo")
            logger.info("redis is ready: " + repr(ip_port))
            return

        except redis.ConnectionError as e:
            logger.info("can not connect to redis: " + repr(ip_port) + " " + repr(e))
            time.sleep(0.1)
            continue
    else:
        logger.error("can not connect to redis: " + repr(ip_port))
        raise

License

The MIT License (MIT) - Copyright (c) 2015 Zhang Yanpo (张炎泼)