Python client

Overview

ruvio-client is the Python client for Ruvio. It has two classes with the same methods: ruvio.Ruvio is synchronous, ruvio.asyncio.Ruvio uses one pipelined connection that all coroutines share. Both have a typed method for every command in the compatibility matrix, plus helpers for locks, rate limits, cache-aside and idempotency.

Install

Install it from the repository:

git clone https://github.com/salihcantekin/ruvio.git
pip install ./ruvio/clients/python             # no dependencies
pip install "./ruvio/clients/python[hiredis]"  # optional: faster reply parsing

A pip install git+https://...#subdirectory=clients/python URL should work the same way; it was not tried. A PyPI release is pending, and until then pip install ruvio-client will not find it. The checks on this page ran with PYTHONPATH pointing at the client directory, not through pip install.

Quick start

Sync client. The connection opens on the first command (lazy=True is the default).

import ruvio

r = ruvio.Ruvio("127.0.0.1", 6379, db=0)        # or ruvio.Ruvio.from_url("ruvio://host:6379/0")
r.set("greeting", "hello", px=60_000)            # => True
r.get("greeting")                                # => 'hello'
r.get("nope")                                    # => None  (nil reply)
r.set("greeting", "hi", ifeq="hello")            # => True  (Ruvio: only if the value matches)
r.incr("visits")                                 # => 1
r.raw.get("visits")                              # => b'1'  (same connection, bytes)
r.close()

Asyncio client. Many coroutines can use the same object at once; they share one socket.

import asyncio
from ruvio.asyncio import Ruvio

async def main():
    async with Ruvio("127.0.0.1", 6379) as r:    # connects lazily if you skip `async with`
        await r.set("k", "v")
        values = await asyncio.gather(*(r.get("k") for _ in range(1000)))
        print(len(values), set(values))          # => 1000 {'v'}

asyncio.run(main())

Constructor options (sync): host, port, db, username, password, client_name, socket_timeout, connect_timeout (5 s), decode_responses (True), unix_socket_path, parser ("auto", "python" or "hiredis"), on_push, lazy, cluster.

The asyncio class has the same options except socket_timeout and lazy; use asyncio.wait_for for a per-call timeout. The password and username options send AUTH on every (re)connect; this was not run here.

How replies look in Python: a bulk string is str (or bytes on client.raw and with decode_responses=False), integers are int, scores are float, nil is None. OK becomes True for commands whose only answer is OK. HGETALL returns a dict, SMEMBERS a set, ZRANGE ... withscores=True a list of (member, score). A string that is not valid UTF-8 comes back with surrogate escapes and round-trips unchanged; use client.raw if you want bytes.

Performance

How the client stays light, from the README and the code:

• One connection is shared. The asyncio client keeps a FIFO queue of pending replies, so concurrent coroutines share the socket. Do not open a connection per request.
• Write coalescing without timers: an idle connection writes a frame at once. While replies are outstanding, frames from the same event-loop iteration are joined into one write (cap 64 KiB). A serial call is never delayed.
• Command names are pre-encoded ($3\r\nGET\r\n). Bulk-length headers and the integers 0 to 65535 come from tables, so they need no int-to-string conversion. GET, SET, INCR, HGET and DEL build their frame with one b"".join.
• Replies are parsed in place. The asyncio side is an asyncio.BufferedProtocol, so the loop reads straight into the parser's reusable bytearray; the sync side uses recv_into on the same kind of buffer. A scalar reply allocates only the value it returns. The parser can resume, so a big array that arrives in chunks is not parsed again from the start.
• If hiredis can be imported, parser="auto" uses it. Force either parser with parser="python" or parser="hiredis" (both were run here).

Measured in the README with benchmarks/bench.py (Python 3.13.16, shared noisy 4-vCPU Linux VM, local server over loopback, 64-byte values, one 2-second run per figure; indicative only, not release evidence):

Parse cost per reply, including the copy into the read buffer (README): +OK about 870 ns pure Python and 610 ns with hiredis, a 64-byte bulk about 1030 and 660 ns, a 10-item bulk array about 5100 and 1900 ns. Encoding a GET takes about 310 ns through the typed method and each GET frame allocates 70 bytes. The comparison with redis-py is the README's own single run on one machine; it is not a general claim.

To go faster: use a pipeline (r.pipeline()) or concurrent coroutines on one asyncio client, reuse the client, install the hiredis extra, and use client.raw when you do not need str decoding.

Commands

Method names follow the command names in lower case (hset, zadd, xadd, json_set, bf_madd, xgroup_create, cluster_slots). The asyncio client has the same names; add await. The calls below were run on a standalone server.

r.mset({"a": 1, "b": 2}); r.mget("a", "b", "zz")        # => ['1', '2', None]
r.set("t", "1"); r.expire("t", 100); r.ttl("t")          # => 100

r.hset("user:1", mapping={"name": "Ada", "age": 36})     # => 2
r.hgetall("user:1")                                      # => {'name': 'Ada', 'age': '36'}

r.rpush("queue", "a", "b", "c"); r.lrange("queue", 0, -1)   # => ['a', 'b', 'c']
r.sadd("tags", "x", "y"); r.sismember("tags", "x")          # => True

r.zadd("board", {"ada": 100, "bob": 50})
r.zrange("board", 0, -1, rev=True, withscores=True)      # => [('ada', 100.0), ('bob', 50.0)]

r.xadd("events", {"type": "signup"})
r.xrange("events")                                       # => [StreamEntry(id='...', fields={'type': 'signup'})]

r.json_set("doc", "$", '{"city":"Izmir"}'); r.json_get("doc")   # => '{"city":"Izmir"}'
r.json_set_obj("u:1", {"name": "Ada", "n": [1, 2]})             # uses the json module
r.json_get_obj("u:1")                                           # => {'name': 'Ada', 'n': [1, 2]}

r.geoadd("places", [(13.361389, 38.115556, "Palermo"), (15.087269, 37.502669, "Catania")])
r.geodist("places", "Palermo", "Catania", "km")          # => 166.27415156960032

r.setbit("flags", 7, 1); r.bitcount("flags")             # => 1
r.pfadd("uniq", "a", "b", "a"); r.pfcount("uniq")        # => 2

Anything else goes through execute_command, which sends the words as written and returns the raw reply:

r.execute_command("STRLEN", "visits")                    # => 1

A Pipeline built from a client has the same methods, so every typed method also works inside a pipeline.

Ruvio commands

These replace Lua scripts or multi-step Redis patterns with one atomic command.

r.set("cfg", "v1")
r.set("cfg", "v2", ifeq="v1")        # => True
r.set("cfg", "v3", ifeq="v1")        # => False, the value is now v2
r.delex("cfg", "v2")                 # => True

r.incrby("seats", 3, max=4)          # => 3
r.incrby("seats", 2, max=4)          # => None: would pass the limit

r.limit("login", 2, 60_000)          # => 1 (remaining)
r.limit("login", 2, 60_000)          # => 0
r.limit("login", 2, 60_000)          # => None: refused

f = r.fillget("page", 10_000)        # => FillResult(state='FILL', value=None, token='15')   # the token is a server counter, your number will differ
r.fillset("page", "<html/>", 60_000, f.token)   # => True
r.fillget("page", 10_000)            # => FillResult(state='HIT', value='<html/>', token=None)

tok = r.lease("job", 30_000)         # => 1 (fencing token)
r.lease("job", 30_000)               # => None while held
r.release("job", tok)                # => True

r.once("pay", 60_000)                # => True
r.once("pay", 60_000)                # => False

r.set("tickets", "5")
r.take("tickets", 2)                 # => 3 left
r.take("tickets", 9)                 # => None: not enough

v1 = r.setv("d", "a")                # => 1
r.setv("d", "b", v1)                 # => 2   (a wrong version raises ResponseError: ERR version mismatch)

r.set("e", "x"); r.getex("e", 5000)  # => 'x', and the key now expires in 5000 ms

r.lpushcap("recent", 3, "1", "2", "3", "4", "5")   # => 3; lrange gives ['5', '4', '3']
r.rpushcap("recent2", 3, "1", "2", "3", "4", "5")  # => 3; lrange gives ['3', '4', '5']

r.bf_reserve("bf", 0.01, 1000)       # => True
r.bf_madd("bf", "a", "b")            # => [True, True]
r.bf_mexists("bf", "a", "z")         # => [True, False]

r.json_index_create("byCity", "user:", "$.city")   # JSON.INDEX CREATE only runs in database 0
r.json_set("user:9", "$", '{"city":"Izmir"}')
r.json_find("byCity", "Izmir")       # => ['user:9']  (while the index builds the server answers an error; retry)

c = r.changes_start()                # => 64485 (a cursor; the number depends on the server)
r.set("note", "x")
r.changes(c)                         # => [(64486, 'SET note')]

r.idem_begin("idem:1", "fp", "me", 60_000)        # => IdemResult(state='acquired', result=None)
r.idem_complete("idem:1", "fp", "me", "ok")       # => 1
r.idem_begin("idem:1", "fp", "other", 60_000)     # => IdemResult(state='completed', result='ok')
r.idem_begin("idem:1", "zz", "other", 60_000)     # => IdemResult(state='conflict', result=None)

A FILLGET that answers STALE comes with a token when you should rebuild, and without one when someone else already does. fillset also takes swr_ms for stale-while-revalidate. fillget(key, fill_ms, wait_ms=...) sends WAIT, which parks the whole shared connection: on the asyncio client run it on client.dedicated().

Helpers

The helpers are methods on the client, on both classes. They only send single-key commands, so they work unchanged with cluster=True (checked, see below). On the asyncio client use async with and await.

# Lock: SET NX PX to take, SET IFEQ to renew, DELEX IFEQ to release
with r.lock("lock:invoice", ttl_ms=10_000) as lk:
    lk.owned()                       # => True
    lk.renew()                       # => True; False means the lock was lost
# after the block the key is gone (get returns None)

lk = r.lock("lock:x", 5000, blocking=False)
lk.acquire()                         # => True
r.lock("lock:x", 5000, blocking=False).acquire()   # => False (held)
lk.release()                         # => True

# Fenced lease and semaphore slot; both RELEASE on exit
with r.lease_scope("lease:job", 30_000) as token: ...        # token => 1
with r.semaphore_scope("sem:export", 4, 30_000) as token: ...  # LockError when full

# Rate limit: (allowed, remaining)
r.rate_limit("rl:ada", 2, 60_000)    # => (True, 1)  then (True, 0)  then (False, 0)

# Cache-aside with stampede protection
page = r.get_or_create("cache:home", lambda: "<home/>", ttl_ms=60_000)
# 5 threads, each with its own client: all got '<home/>', the factory ran once

# Idempotency
r.idempotent("idem:pay", "fp", 60_000, lambda: "ok")      # => ('executed', 'ok')
r.idempotent("idem:pay", "fp", 60_000, lambda: "again")   # => ('completed', 'ok')
r.idempotent("idem:pay", "other", 60_000, lambda: "x")    # => ('conflict', None)

# Lazy scans: scan_iter, hscan_iter, sscan_iter, zscan_iter
sorted(r.scan_iter(match="user:*", count=2))   # => ['user:0', ..., 'user:4']
dict(r.hscan_iter("h"))                        # => {'a': '1', 'b': '2'}

# Typed JSON with the standard json module
r.json_set_obj("u:1", {"name": "Ada", "age": 36})
r.json_get_obj("u:1")                          # => {'name': 'Ada', 'age': 36}
r.json_mget_obj(["u:1", "nope"])               # => [{'name': 'Ada', 'age': 36}, None]

# Recent items and "seen before"
r.push_recent("recent:ada", "item-1", 3); r.push_recent("recent:ada", "item-2", 3)
r.lrange("recent:ada", 0, -1)                  # => ['item-2', 'item-1']
r.seen_before("seen:emails", "a@x", "b@x")     # => [False, False]  (and records them)
r.seen_before("seen:emails", "a@x", "c@x")     # => [True, False]

The same on asyncio (run):

async with r.lock("lock:invoice", ttl_ms=10_000) as lk:
    await lk.owned(); await lk.renew()           # => True, True
async with r.lease_scope("lease:job", 30_000) as token: ...

async def build():
    await asyncio.sleep(0.05); return "<home/>"
await asyncio.gather(*(r.get_or_create("cache:home", build, 60_000) for _ in range(10)))
# => ten '<home/>', the factory ran once

async def work(): return "ok"
await r.idempotent("idem:pay", "fp", 60_000, work)     # => ('executed', 'ok')

keys = [k async for k in r.scan_iter(match="user:*", count=2)]

Failure behavior:

• Lock. acquire returns False when the lock is held and blocking=False. The with form raises ruvio.LockError when it cannot get the lock (with timeout=0.2 it retried every retry_ms and raised after 0.2 s). There is no automatic renewal: call renew() yourself, False means you no longer own it. Renew and release compare the token, so they never touch another holder's lock. If a release fails, the key expires by its TTL.
• Scopes. lease_scope and semaphore_scope raise LockError when the lease is held or the semaphore is full.
• get_or_create. Waiters poll plain FILLGET with a backoff from 5 ms up to 100 ms, never WAIT. After timeout (10 s by default) a waiter raises ruvio.TimeoutError. If your factory raises, the exception reaches you and the fill marker lapses after fill_ms.
• Rate limit and idempotency. Server errors and timeouts raise. Nothing is retried for you; an unknown outcome after a timeout stays unknown. A caller that sees ("pending", None) from idempotent should wait and ask again.

Pipelines, transactions, Pub/Sub

Concurrent coroutines on one asyncio client are already pipelined. For an explicit group, use a pipeline: one write, replies in order.

p = r.pipeline()                       # asyncio: await p.execute()
p.set("a", 1).incr("a").get("a")
p.execute()                            # => [True, 2, '2']

t = r.pipeline(transaction=True)       # MULTI ... EXEC in one write
t.set("x", "1").append("x", "2")
t.execute()                            # => [True, 2]

# WATCH + retry (sync); the function may run more than once
r.set("bal", 10)
r.transaction(lambda pipe: pipe.set("bal", int(r.get("bal")) - 1), "bal")   # => [True], bal is now 9

# asyncio: transaction() opens a dedicated connection, because WATCH applies to the whole connection
async def dec(pipe):
    pipe.set("bal", int(await r.get("bal")) - 1)
await r.transaction(dec, "bal")        # => [True]

# errors inside a pipeline
p = r.pipeline(); p.set("s", "v").incr("s").get("s")
p.execute(raise_on_error=False)        # => [True, ResponseError('ERR value is not an integer or out of range'), 'v']

execute() raises the first error by default; raise_on_error=False returns the exceptions in their places. An unknown command queued inside MULTI makes EXEC answer EXECABORT for the whole block and run nothing, as Redis does; the client raises ExecAbortError. Older server builds ran the other commands and only the refused one failed.

Pub/Sub uses its own socket:

ps = r.pubsub(ignore_subscribe_messages=True)    # asyncio: ps = await r.pubsub()
ps.subscribe("news")                             # also psubscribe("ev.*"), ssubscribe("chan")
r2.publish("news", "hello")                      # from another client or thread
for msg in ps.listen():                          # asyncio: async for msg in ps.listen()
    print(msg.kind, msg.channel, msg.data, msg.pattern)   # => message news hello None
    break
ps.close()

If the socket drops, the next get_message reopens it and subscribes again. r.hello(3) switches a connection to RESP3 (the call returned the server info map; the typed converters accept RESP3 maps and doubles); out-of-band pushes such as CLIENT TRACKING invalidations go to on_push=callback and never reach a command's reply.

Blocking commands (blpop, xread with block, fillget with wait) hold the connection. On the sync client that only blocks you; on asyncio use dedicated():

d = r.dedicated()                      # new client, new connection, same settings
await d.blpop("nolist", timeout=0.1)   # => None on timeout
await d.close()

Cluster

Pass cluster=True and one seed address (any shard). The plain class and its fast path are not touched: with the flag, the constructor returns a ClusterRuvio (sync or asyncio). On the first command it sends CLUSTER SLOTS, builds a 16384-entry slot table and opens one connection per node on demand.

A standalone server answers -ERR cluster mode is disabled; the client then behaves as a normal standalone client. is_cluster tells which (a property on sync, await client.is_cluster() on asyncio). The slot is CRC16 of the key or its {hash tag}; ruvio.key_slot(key) gives it.

• Single-key commands go to the node that owns the slot. Keyless ones go to the seed node.
• MGET, MSET, DEL, UNLINK, EXISTS, TOUCH across slots are split per node, sent to all nodes before any reply is read, and put back in the caller's order. DBSIZE is summed, KEYS is concatenated, FLUSHDB runs on every node.
• Other multi-key commands (SUNION, RENAME, BITOP, LMOVE, ...) route by their real keys. If the keys are in different slots, the client raises ruvio.CrossSlotError before sending anything.
• MOVED patches the slot table (or re-reads CLUSTER SLOTS when more is wrong) and the command is retried once on the node named. A second MOVED raises MovedError.
• Pipelines are grouped per node, each group goes out as one write, replies come back in the caller's order. Transactions need all keys in one slot, else CrossSlotError. MULTI/EXEC sent one by one are refused: use pipeline(transaction=True).
• SCAN walks the nodes one after the other; the cursor is opaque. scan_iter() does it for you. Only database 0 exists in cluster mode.

import ruvio

c = ruvio.Ruvio("127.0.0.1", 17450, cluster=True)   # any shard port
c.is_cluster                                         # => True
ruvio.key_slot("somekey")                            # => 11058, same as CLUSTER KEYSLOT
c.set("session:ada", "hello"); c.set("somekey", "other shard")
c.mget("session:ada", "somekey", "missing")          # => ['hello', 'other shard', None]

keys = [f"k{i}" for i in range(60)]
c.mset({k: i for i, k in enumerate(keys)})           # split over the shards
c.mget(*reversed(keys))[:3]                          # => ['59', '58', '57']  (caller's order)

p = c.pipeline()
for i in range(10): p.set(f"p{i}", i)
for i in range(10): p.get(f"p{i}")
p.execute()[10:]                                     # => ['0', '1', ..., '9']

t = c.pipeline(transaction=True)
t.set("{u}:a", 1).incr("{u}:a"); t.execute()         # => [True, 2]  same hash tag, same slot

c.sunion("a", "somekey")
# => CrossSlotError: CROSSSLOT SUNION keys don't hash to the same slot (nothing was sent)

with c.lock("lock:invoice", 10_000): ...             # helpers work unchanged
c.get_or_create("cache:home", lambda: "<home/>", 60_000)

Asyncio (run on the same 3-shard server; 300 concurrent set calls, then a reversed mget):

from ruvio.asyncio import Ruvio

async with Ruvio("127.0.0.1", 17450, cluster=True) as a:
    await a.is_cluster()                                         # => True
    await asyncio.gather(*(a.set(f"a{i}", i) for i in range(300)))
    (await a.mget(*[f"a{i}" for i in range(300)][::-1]))[:3]     # => ['299', '298', '297']
    d = a.dedicated()                                            # blocking calls: own connection to the owner node
    await d.blpop("nolist", timeout=0.1)                         # => None
    await d.close()

Standalone fallback, run against the plain server: ruvio.Ruvio("127.0.0.1", 17394, cluster=True) returned a working client and is_cluster was False. A spread check: 60 keys written with one mset ended up on all three shards (21, 22 and 33 keys when counted over direct connections: 76 keys in all, the 60 from mset plus keys from the other examples; dbsize() through the cluster client returned 76, the sum).

The README's own cluster benchmark (one run, same noisy VM, not release evidence) shows the cluster client at about half the standalone throughput in the CPU-bound asyncio case (64 concurrent GET: 239k standalone against 129k cluster ops/s), because routing runs in Python and batches per connection are smaller.

Errors, nil, reconnect, cancellation

• A server error raises ruvio.ResponseError or a subclass: WrongTypeError, NoScriptError, NoGroupError, BusyError, ExecAbortError, MovedError. Each has .code (such as "WRONGTYPE"). The connection stays usable. Cluster adds CrossSlotError; helpers raise LockError, WatchError. Socket problems raise ruvio.ConnectionError or ruvio.TimeoutError. All derive from ruvio.RuvioError.
• Nil maps to None. Commands that "refuse" (limit, take, lease, incrby ... max) return None.
• No command is sent again after a connection loss. The command that sees the drop raises ConnectionError; the next command reconnects and sends AUTH, SELECT and CLIENT SETNAME again.
• Timeouts: sync uses socket_timeout. Asyncio has no such option; wrap the call in asyncio.wait_for. Cancelling an awaiting caller is safe: the late reply is dropped and the connection keeps working.

r.set("greet", "hi")
r.incr("greet")            # ResponseError: ERR value is not an integer or out of range
r.lpush("greet", "x")      # WrongTypeError (code 'WRONGTYPE')
r.execute_command("NOSUCH")  # ResponseError: ERR unknown command
ruvio.Ruvio("127.0.0.1", 17999).get("k")   # ConnectionError: cannot connect to 127.0.0.1:17999: [Errno 111] Connection refused

# sync timeout
slow = ruvio.Ruvio("127.0.0.1", 6379, socket_timeout=0.2)
slow.blpop("nolist", timeout=2)            # TimeoutError: timed out waiting for the reply

# asyncio timeout and cancellation (run)
try: await asyncio.wait_for(r.get("k"), 0.0000001)
except asyncio.TimeoutError: pass
await r.ping()                             # => 'PONG'  the connection still works
t = asyncio.ensure_future(r.get("k")); await asyncio.sleep(0); t.cancel()
await r.ping()                             # => 'PONG'

Not done / not verified

• See PyPI for the release status.
• Threads. The sync client is not thread-safe and has no connection pool. Use one client per thread (the get_or_create check above did that) or dedicated().
• Typing. Command methods are shared by the sync client, the async client and pipelines, so return annotations are Any. Each docstring states the shape of the value.
• Asyncio write backpressure. The client does not handle pause_writing. Writes are capped at 64 KiB each, but the transport buffer can grow without limit if the server stops reading.
• No TLS option. RESP3 is available through hello(3); with hiredis, RESP3 big numbers come back as str (pure Python returns int).
• Cluster. Routing is live-tested on 3 shards. Refused rather than emulated: cross-slot RENAME, COPY, SMOVE, LMOVE. CLUSTER NODES/SHARDS/INFO/MYID and CHANGES go to the seed only. No ASK handling, no replica reads. Not verified: a topology that changes while the client runs (MOVED recovery was tested by corrupting the client's table, not by moving slots), clusters with more than 3 nodes, AUTH on a cluster, a pipeline above 64 KiB per node with a dead peer.
• Performance figures are single noisy runs from the README, not repeated here, and the redis-py comparison is one run on one machine.
• Not run for this page: username/password, unix_socket_path, ACL SAVE/LOAD end to end (needs security.acl_file).

Testing

cd clients/python
python -m venv .venv && .venv/bin/pip install pytest pytest-asyncio     # optional: hiredis redis
.venv/bin/python -m pytest                                              # no server: codec and cluster unit tests
RUVIO_TEST_ADDR=127.0.0.1:6379 .venv/bin/python -m pytest              # + live tests (database 11, runs FLUSHDB)
RUVIO_TEST_ADDR=127.0.0.1:6379 RUVIO_CLUSTER_ADDR=127.0.0.1:6380 .venv/bin/python -m pytest   # + cluster
RUVIO_TEST_PARSER=python ...                                            # force the pure-Python parser

219 tests are collected: test_resp.py 16 (encoder, both parsers, byte-by-byte feeding, RESP3 types), test_live.py 54 (one scenario per command family, each on the sync and the asyncio client), test_helpers_live.py 23 (helpers, Pub/Sub, RESP3 pushes, multiplexing order, cancellation, reconnect, timeouts, large values, WATCH), test_coverage.py 3 (the 287-command check), test_cluster_unit.py 86 (slots, hash tags, key extraction, a routing rule for all 287 commands, MOVED handling and fail-fast CROSSSLOT against scripted fake servers) and test_cluster_live.py 37 (3-shard server: slot table against CLUSTER KEYSLOT, spread, split-command order, stale-table MOVED recovery, transactions, helpers, scans, pub/sub, losing one node).

Results from my run on 12 October 2026, with the standalone server on port 17394 and the 3-shard server on 17450 to 17452: 217 passed, 2 skipped, with the default parser and again with RUVIO_TEST_PARSER=python (same counts). Without a server: 105 passed, 114 skipped.

The two skips are the FLUSHALL test, once on the sync and once on the asyncio client; it needs RUVIO_TEST_FLUSHALL=1.

For three shards, set shards = 3 and port_mode = "per-shard" in the config; shard i listens on base port + i. See Start for starting a server.

See also: Clients overview, Rust, C#, Go, Java, Node.js, Examples, Commands, Packages.