WRedis

WRedis — Production-ready Python library for Redis with sync/async APIs, cache decorators, high availability, and comprehensive type safety.

Installation

pip install wredis

Requires Python 3.9+ and a running Redis server (local or remote).

Features

Quick Start

from wredis.cache import RedisCacheManager

cache = RedisCacheManager(host="localhost")

@cache("my_cache", ttl=60)
def expensive_function(param: str) -> dict:
    return {"result": f"processed {param}"}

data = expensive_function("hello")  # cached for 60s

Modules

Bitmap

from wredis.bitmap import RedisBitmapManager

bm = RedisBitmapManager(host="localhost")
bm.set_bit("my_bitmap", offset=5, value=1)
print(bm.get_bit("my_bitmap", 0))   # 0
print(bm.count_bits("my_bitmap"))    # 1

Methods: set_bit, get_bit, exist, count_bits, get_ttl, extend_ttl, delete_bitmap

Hash

from wredis.hash import RedisHashManager

hm = RedisHashManager(host="localhost")
hm.create_hash("users", "user:1", {"name": "Alice", "age": 30}, ttl=60)
user = hm.read_hash("users", "user:1")
all_users = hm.read_all_hash("users")
hm.update_hash("users", "user:1", {"age": 31})
hm.delete_hash_field("users", "user:2")
hm.delete_hash("users")
exists = hm.exist("users")

Methods: create_hash, read_hash, exist, update_hash, delete_hash_field, delete_hash, read_all_hash, get_ttl, extend_ttl

Queue

from wredis.queue import RedisQueueManager

# Producer
qm = RedisQueueManager(host="localhost")
qm.publish("tasks", {"id": 1, "task": "process_image"}, ttl=30)
exists = qm.exist("tasks")

Methods: publish, on_message, exist, start, stop, wait, get_queue_length, delete_queue

Pub/Sub

from wredis.pubsub import RedisPubSubManager

pbm = RedisPubSubManager(host="localhost")

# Producer
pbm.publish_message("notifications", "Hello, Redis!")
pbm.publish_message("alerts", {"severity": "high", "message": "Disk full"})
exists = pbm.exist("notifications")

# Consumer
@pbm.on_message("notifications")
def handle(msg):
    print(f"Received: {msg}")

pbm.stop_listeners()

Methods: publish_message, on_message, exist, stop_listeners

Streams

from wredis.streams import RedisStreamManager

sm = RedisStreamManager(host="localhost")

# Producer
sm.add_to_stream("events", {"action": "login", "user": "alice"})
exists = sm.exist("events")

# Consumer
@sm.on_message("events", group_name="my_group", consumer_name="worker_1")
def process(data):
    print(f"Processing: {data}")

sm.wait()

Methods: add_to_stream, on_message, exist, read_from_stream, wait, delete_stream

Sets

from wredis.sets import RedisSetManager

sm = RedisSetManager(host="localhost")
sm.add_to_set("tags", "python", "redis", "wredis")
members = sm.get_set_members("tags")
print(sm.is_member("tags", "python"))  # True
sm.remove_from_set("tags", "redis")
exists = sm.exist("tags")

Methods: add_to_set, get_set_members, exist, is_member, remove_from_set, get_ttl, extend_ttl, delete_set

Sorted Sets

from wredis.sortedset import RedisSortedSetManager

ssm = RedisSortedSetManager(host="localhost")
ssm.add_to_sorted_set("leaderboard", 100, "player1")
ssm.add_to_sorted_set("leaderboard", 200, "player2")
ssm.add_to_sorted_set("leaderboard", 150, "player3")

top = ssm.get_sorted_set_reverse("leaderboard", with_scores=True)
rank = ssm.get_rank("leaderboard", "player2")
score = ssm.get_score("leaderboard", "player3")
ssm.increment_score("leaderboard", 50, "player3")
exists = ssm.exist("leaderboard")

Methods: add_to_sorted_set, get_sorted_set, get_sorted_set_reverse, exist, remove_from_sorted_set, get_rank, get_score, delete_sorted_set, set_ttl, get_ttl, increment_score, get_sorted_set_by_score

Geo

from wredis.geo import RedisGeoManager

gm = RedisGeoManager(host="localhost")
gm.add_location("places", "Central Park", 40.785091, -73.968285)
dist = gm.distance("places", "Central Park", "Times Square")
exists = gm.exist("places")

Methods: add_location, exist, distance, geo_radius, get_location, delete_geo

Advanced Patterns

For complex use cases, WRedis supports:

• Pipelines & Transactions: Atomic operations and batch execution.
• Custom Key Generation: Full control over cache key naming.
• Connection Tuning: Optimization for high-traffic environments.
• Granular Error Handling: Specific exception hierarchy for robust applications.

Check the Full Documentation for detailed guides.

HyperLogLog

from wredis.hyperloglog import RedisHyperLogLogManager

hll = RedisHyperLogLogManager(host="localhost")
hll.add("visitors", "user1", "user2", "user3")
count = hll.count("visitors")
hll.merge("all_visitors", "visitors")
exists = hll.exist("visitors")

Methods: add, exist, count, merge, get_all, delete_hyperloglog

Pipeline

from wredis.pipeline import RedisPipelineManager

pm = RedisPipelineManager(host="localhost")
pm.set_get("key1", "value1")
results = pm.mget_pipeline("key1", "key2", "key3")

Methods: set_get, mget_pipeline

Transaction

from wredis.transaction import RedisTransactionManager

tm = RedisTransactionManager(host="localhost")
result = tm.get_and_set("counter", "1")

Methods: get_and_set

Cache Decorator

from wredis.cache import RedisCacheManager

cache = RedisCacheManager(host="localhost")

@cache(ttl=60)
def get_user(user_id: str) -> dict:
    return {"id": user_id, "name": "Alice"}

# Call decorated function — results are cached
user = get_user("42")

Methods: @cache (sync), @async_cache (async), invalidate, clear, get_stats

High Availability

Sentinel

from wredis.ha.sentinel import RedisSentinelManager

sm = RedisSentinelManager(
    sentinel_hosts=[("sentinel1", 26379), ("sentinel2", 26379)],
    service_name="mymaster",
    password="optional",
)

Cluster

from wredis.ha.cluster import RedisClusterManager

cm = RedisClusterManager(
    startup_nodes=[{"host": "node1", "port": 6379}],
    password="optional",
)

Configuration

All managers accept common connection parameters:

Queue-specific:

Testing

# Run all tests
pytest tests/

# Run with coverage
pytest --cov=wredis tests/

# Run specific module tests
pytest tests/unit/test_queue.py
pytest tests/integration/test_all_modules_integration.py

License

MIT — see LICENSE for details.