Redis Reliability Patterns

import redis
import time

r = redis.Redis()

# Pattern 1: Fixed window counter
def is_rate_limited_fixed(user_id, limit=100, window=60):
    key = f"rate:{user_id}:{int(time.time()) // window}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)
    return count > limit

# Pattern 2: Sliding window (more accurate)
def is_rate_limited_sliding(user_id, limit=100, window=60):
    key = f"rate:{user_id}"
    now = time.time()
    pipe = r.pipeline()
    pipe.zremrangebyscore(key, 0, now - window)  # remove old entries
    pipe.zadd(key, {str(now): now})              # add current request
    pipe.zcard(key)                               # count requests
    pipe.expire(key, window)                      # set expiry
    results = pipe.execute()
    return results[2] > limit

# Pattern 3: Token bucket (smooth rate limiting)
def consume_token(user_id, capacity=100, refill_rate=10):
    key = f"bucket:{user_id}"
    now = time.time()

    tokens = r.hgetall(key)
    if not tokens:
        r.hset(key, mapping={"tokens": capacity - 1, "last_refill": now})
        r.expire(key, 300)
        return True  # allowed

    last_refill = float(tokens[b"last_refill"])
    current_tokens = float(tokens[b"tokens"])

    # Refill tokens based on elapsed time
    elapsed = now - last_refill
    current_tokens = min(capacity, current_tokens + elapsed * refill_rate)

    if current_tokens >= 1:
        r.hset(key, mapping={"tokens": current_tokens - 1, "last_refill": now})
        return True  # allowed
    return False  # rate limited

import uuid
import time
import redis

r = redis.Redis()

class RedisLock:
    def __init__(self, name, ttl=10):
        self.name = f"lock:{name}"
        self.ttl = ttl
        self.token = str(uuid.uuid4())

    def acquire(self, timeout=10):
        end = time.time() + timeout
        while time.time() < end:
            if r.set(self.name, self.token, nx=True, ex=self.ttl):
                return True
            time.sleep(0.01)
        return False

    def release(self):
        # Lua script ensures atomic check-and-delete
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        return r.eval(script, 1, self.name, self.token)

# Usage
lock = RedisLock("process_order:123")
if lock.acquire():
    try:
        # Process the order
        process_order(123)
    finally:
        lock.release()

# Basic pub/sub
SUBSCRIBE channel:user:1
PUBLISH channel:user:1 '{"event": "login", "user_id": 1}'

# Pattern subscription
SUBSCRIBE channel:order:*
PUBLISH channel:order:created '{"order_id": 42}'

# Redis Streams as durable pub/sub (recommended)
XADD user_events * user_id 1 event login
XADD user_events * user_id 2 event purchase

# Consumer group for parallel processing
XGROUP CREATE user_events processors 0
XREADGROUP GROUP processors worker1 COUNT 10 STREAMS user_events >

import redis
import json
import time

r = redis.Redis()

def enqueue_job(queue_name, job_data):
    """Add job to queue with metadata"""
    job = {
        "id": str(uuid.uuid4()),
        "data": job_data,
        "created_at": time.time(),
        "attempts": 0
    }
    r.rpush(f"queue:{queue_name}", json.dumps(job))
    return job["id"]

def dequeue_job(queue_name, timeout=30):
    """Blocking dequeue with visibility timeout"""
    result = r.blpop(f"queue:{queue_name}", timeout=timeout)
    if result is None:
        return None

    job = json.loads(result[1])
    job["attempts"] += 1

    # Move to processing queue with TTL
    r.setex(
        f"processing:{job['id']}",
        300,  # 5 minute visibility timeout
        json.dumps(job)
    )
    return job

def complete_job(job_id):
    """Mark job as complete"""
    r.delete(f"processing:{job_id}")

def requeue_job(job_id):
    """Requeue a failed job"""
    job_data = r.get(f"processing:{job_id}")
    if job_data:
        job = json.loads(job_data)
        if job["attempts"] < 3:
            r.rpush(f"queue:default", json.dumps(job))
        r.delete(f"processing:{job_id}")

def process_payment(user_id, amount, idempotency_key):
    """Process payment with idempotency protection"""
    lock_key = f"idempotent:{idempotency_key}"

    # Check if operation already completed
    result = r.get(lock_key)
    if result:
        return json.loads(result)  # return cached result

    # Try to claim this idempotency key
    claimed = r.set(lock_key, "processing", nx=True, ex=3600)
    if not claimed:
        # Another request is processing this key
        time.sleep(0.1)
        return process_payment(user_id, amount, idempotency_key)

    try:
        # Process the payment
        payment_result = charge_payment(user_id, amount)

        # Store the result
        r.setex(lock_key, 3600, json.dumps(payment_result))
        return payment_result
    except Exception as e:
        r.delete(lock_key)  # allow retry on failure
        raise

class CircuitBreaker:
    def __init__(self, name, failure_threshold=5, recovery_timeout=60):
        self.name = f"circuit:{name}"
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout

    def is_open(self):
        state = r.hgetall(self.name)
        if not state:
            return False

        if state[b"state"] == b"open":
            # Check if recovery period has elapsed
            opened_at = float(state[b"opened_at"])
            if time.time() - opened_at > self.recovery_timeout:
                r.hset(self.name, "state", "half-open")
                return False
            return True
        return False

    def record_success(self):
        r.delete(self.name)

    def record_failure(self):
        failures = r.hincrby(self.name, "failures", 1)
        if failures >= self.failure_threshold:
            r.hset(self.name, mapping={
                "state": "open",
                "opened_at": time.time()
            })