redi.py

A pure Python distributed data structures library, equivalent to Redisson for Python. Fully compatible with Redisson at the data structure and encoding level. Supports Redis, Redis Cluster, and Redis Sentinel.

Features

• 67+ distributed data structures — collections, locks, atomics, caches, pub/sub, Redis Stack
• Redisson wire-compatible — exact Lua scripts, HighwayHash, struct-pack, prefixName logic
• Full Async API — AsyncRedissonClient with 67+ async structures via redis.asyncio
• Pythonic API — MutableMapping, MutableSequence, MutableSet, context managers, numeric protocols
• JSON codec — same encoding as redi.php, readable by Redisson (JsonJacksonCodec) with Long wrapping
• Lua scripts for atomicity — lock safety, CAS, semaphore, rate limiting, bloom filter
• redis-py native — leverages built-in ConnectionPool for thread-safe operations
• Background threads — watchdog renewal, delayed queue migration, map cache eviction
• Pipeline + Transaction — RBatch and RTransaction for atomic operations
• Cluster + Sentinel — RedissonClusterClient and RedissonSentinelClient
• pytest tested — 521 tests covering all data structures

Quick Start

pip install redi.py

from redipy import RedissonClient

client = RedissonClient()          # auto-connects to localhost:6379

# RMap (works like a Python dict!)
m = client.get_map("myMap")
m["key"] = "value"
print(m["key"])                    # "value"
print(len(m))                      # 1
print("key" in m)                  # True
for k in m:                        # iterates keys
    print(k, m[k])

# RList (works like a Python list!)
lst = client.get_list("myList")
lst.append("a")
lst.append("b")
print(lst[0])                      # "a"
print(len(lst))                    # 2

# RSet (works like a Python set!)
s = client.get_set("mySet")
s.add("x")
print("x" in s)                    # True

# RLock (use with context manager!)
with client.get_lock("myLock"):
    # critical section - auto unlock on exit
    pass

# RSemaphore (context manager too)
with client.get_semaphore("mySem", permits=3):
    pass

# RRateLimiter
rl = client.get_rate_limiter("myLimiter")
rl.set_rate(5, 1000)  # 5 tokens per second
rl.try_acquire()  # True

client.shutdown()

Supported Data Structures

Category	Structures
Collections	RMap, RList, RSet, RSortedSet, RQueue, RDeque
Blocking	RBlockingQueue, RBlockingDeque
Priority	RPriorityQueue, RPriorityDeque
Delayed	RDelayedQueue, RTransferQueue, RCircularBuffer
Locks	RLock, RReadWriteLock, RFairLock, RMultiLock, RRedLock, RFencedLock
Synchronization	RSemaphore, RCountDownLatch, RPermitExpirableSemaphore
Atomics	RAtomicLong, RAtomicDouble, RLongAdder, RDoubleAdder
Cache	RMapCache, RSetCache, RLocalCachedMap, RBucket, RBuckets
Bitmap	RBitSet, RBitVectorStore, RBloomFilter
Pub/Sub	RTopic, RPatternTopic, RShardedTopic, RReliableTopic
Rate Limiting	RRateLimiter
Specialized	RHyperLogLog, RGeo, RStream, RTimeSeries
Sorted	RSortedSet, RScoredSortedSet, RLexSortedSet
ID Generation	RIdGenerator
Scripting	RScript
Transaction	RBatch, RTransaction
Binary	RBinaryStream
Redis Stack	RJsonBucket, RSearch, RTDigest, RTopK, RCuckooFilter, RFunction, RVectorSet
Key Mgmt	RKeys
Executor	RExecutorService, RRemoteService
Clients	RedissonClient, RedissonClusterClient, RedissonSentinelClient, AsyncRedissonClient

Configuration

# Standard Redis
client = RedissonClient({
    "host": "redis.example.com",
    "port": 6379,
    "password": "secret",
    "db": 0,
})

# Redis Cluster
client = RedissonClusterClient({
    "host": "cluster.example.com",
    "port": 7000,
})

# Redis Sentinel
client = RedissonSentinelClient({
    "sentinels": ["sentinel1:26379", "sentinel2:26379"],
    "master_name": "mymaster",
})

Or via environment variables: REDIS_HOST, REDIS_PORT, REDIS_PASSWORD, REDIS_DB.

Async API

import asyncio
from redipy import AsyncRedissonClient

async def main():
    # Context manager auto-connects and disconnects
    async with AsyncRedissonClient() as client:
        # AsyncMap
        m = client.get_map("myMap")
        await m.put("key", "value")
        v = await m.get("key")
        
        # AsyncLock (async context manager)
        async with client.get_lock("myLock") as lock:
            await lock.try_lock(wait_time=1000, lease_time=10000)
        
        # AsyncAtomicLong
        counter = client.get_atomic_long("counter")
        new_val = await counter.increment_and_get()
        
        # AsyncBucket
        bucket = client.get_bucket("myBucket")
        await bucket.set({"data": 42})
        print(await bucket.get())

asyncio.run(main())

Interoperability

All values are stored as JSON strings in Redis, using Redisson-compatible key naming and wire formats:

• RMap("mymap") → Redis HASH key mymap, field/value = JSON encoded
• RLock("mylock") → Redis HASH with field {client_id}:{thread_id}, channel redisson_lock__channel:{mylock}
• RBloomFilter("bf") → Redis bitmap with HighwayHash 128-bit, config at {bf}:config
• RMapCache("mc") → HASH with struct-packed values, ZSET redisson__timeout__set:{mc}
• RSetCache("sc") → Single ZSET with score=expiry_ms, member=JSON value
• RAtomicLong("al") → Redis STRING "42" (decimal, StringCodec compatible)
• RLock unlock → PUBLISH on channel redisson_lock__channel:{name} with message 0

See COMPATIBILITY.md for the full wire-format reference.

Running Tests

pytest -q    # 521 tests, requires redis on localhost:6379