redis-rs 0.15.0

Python wrapper for redis-rs

Python wrapper for:

redis-rs,

bb8,

bb8-redis,

Features

• Async client for single and cluster

• Support typing

• Encoding values from str, int, float

• Decoding values to str, int, float, list, dict

• Partial implementation redis mock-client for testing mode

Install

pip install redis-rs

Using

import asyncio
import redis_rs


async def main():
    async with redis_rs.create_client(
        "redis://redis-node001",
        "redis://redis-node002",
        max_size=1,
        cluster=True,
        features=[
          "mock",  # enable implementation for testing
        ],
    ) as x:
        info = await x.execute("INFO", "SERVER", encoding="info")
        print(info["redis_version"])

        # parse value as json
        print(await x.get("foo", encoding="json"))

        print(await x.execute(b"HSET", "fooh", "a", b"asdfg"))
        print(await x.fetch_int("HSET", "fooh", "b", 11234567890))
        print(await x.fetch_int("HGET", "fooh", "b"))
        print(await x.fetch_str("HGET", "fooh", "a"))
        print(await x.fetch_dict("HGETALL", "fooh", encoding="utf-8"))
        print(await x.hgetall("fooh", encoding="utf-8"))
        print(await x.execute("CLUSTER", "NODES"))
        print(await x.fetch_bytes("GET", "foo"))
        print(await x.fetch_int("GET", "foo"))
        print(await x.execute("HGETALL", "fooh"))
        print(await x.execute("ZADD", "fooz", 1.5678, "b"))
        print(await x.fetch_scores("ZRANGE", "fooz", 0, -1, "WITHSCORES"))
        print(x.status())

        stream = "redis-rs"
        print("x.xadd", await x.xadd(stream, "*", {"a": "1234", "d": 4567}))
        print("x.xadd", await x.xadd(stream, items={"a": "1234", "d": 4567}))
        print("x.xadd", await x.xadd(stream, {"a": "1234", "d": 4567}))
        print("x.xadd", await x.xadd(stream, "*", "a", "1234", "d", 4567))
        print("x.xadd", await x.xadd(stream, "a", "1234", "d", 4567))
        print("xadd", await x.fetch_str("XADD", stream, "*", "a", "1234", "d", 4567))
        print("xread", await x.execute("XREAD", "STREAMS", stream, 0))
        print("xread", await x.fetch_dict("XREAD", "STREAMS", stream, 0, encoding="int"))
        print("x.xread", await x.xread({stream: 0}, encoding="int"))
        print("x.xread", await x.xread(stream, id=0, encoding="int"))
        print("x.xread", await x.xread(stream, stream))


asyncio.run(main())

Development

cargo fmt
cargo clippy
maturin develop

or use hatch envs:

hatch run fmt
hatch run check
hatch run build