redis-infra

Safe, typed, reusable production patterns on top of redis-py — not a thin wrapper. Every capability turns a Redis primitive into application infrastructure you can drop in without re-deriving the correctness details (atomicity, clock skew, key namespacing, delivery guarantees).

Matching sync and async APIs across the board. Python 3.11+.

Install

pip install redis-infra          # core
pip install "redis-infra[orjson]"  # faster JSON

What's in the box

Quick start

from redis_infra import RedisConfig, redis_client, Cache, RateLimiter

client = redis_client(RedisConfig(host="localhost", port=6379, namespace="app"))

# Cache with a compute-on-miss factory (stampede-guarded)
cache = Cache(client, namespace="app")
user = cache.get_or_set("user:1", lambda: load_user(1), ttl=300, stampede=True)

# Atomic rate limit — being limited is a normal result, not an exception
limiter = RateLimiter(client, limit=100, window=60)  # 100 / 60s, sliding window
r = limiter.allow("user:1")
if not r.allowed:
    raise TooManyRequests(retry_after=r.retry_after)

Async mirrors it exactly:

from redis_infra import async_redis_client, AsyncCache

client = async_redis_client(RedisConfig(host="localhost"))
cache = AsyncCache(client, namespace="app")
user = await cache.get_or_set("user:1", load_user_async, ttl=300)

Durable queue (Redis Streams)

from redis_infra import StreamQueue

q = StreamQueue(client, "jobs", group="workers", consumer="w1")
q.ensure_group()
q.publish({"task": "resize", "id": 42})


def handle(msg):
    do_work(msg.data)  # MUST be idempotent — at-least-once delivery


q.run(handle, should_stop=lambda: shutting_down)  # ack on success, DLQ on poison

Guarantees & non-guarantees

• Rate limiting is atomic. Each decision is one server-side Lua script reading the Redis TIME clock — correct across many app instances and clock skew. No Python read-modify-write.
• burst is token-bucket only. Only that algorithm has a bucket capacity to raise, so passing burst to fixed_window or sliding_window raises RateLimitError instead of being quietly ignored.
• Queue is at-least-once, never exactly-once. A message stays pending until acked; a crashed worker's in-flight messages are reclaimed (XAUTOCLAIM) and redelivered. Handlers must be idempotent. Poison messages (redelivered past max_deliveries) are dead-lettered.
• Pub/Sub is not durable. Messages published with no subscriber connected are lost. Use StreamQueue when you need delivery.
• Locks are single-node. RedisLock is a token-owned SET NX PX lock with Lua compare-and-delete release — safe against releasing someone else's lock, and auto-expiring so a crash can't deadlock. It is not Redlock; don't rely on it across independent masters.
• Errors are never swallowed. redis-py errors are wrapped in this package's exception tree (RedisInfraError and subclasses) with __cause__ preserved.

Configuration

RedisConfig is frozen/validated. Build from code or the environment:

RedisConfig.from_env()  # REDIS_URL, REDIS_HOST, REDIS_PORT, REDIS_NAMESPACE, REDIS_SSL, ...

No credentials are ever read from source — supply them via RedisConfig or env vars.

Development

uv pip install -e ".[dev,orjson]"
ruff check . && ruff format --check .
mypy src
pytest tests/unit            # pure, no Redis needed
pytest -m integration        # requires a live Redis (see repo docker-compose.yml)

License

MIT