Cache Transactions: Atomic Operations with MULTI, EXEC, and WATCH

import redis
import time

r = redis.Redis(decode_responses=True)

class CacheTransaction:
    def __init__(self, redis_client):
        self.r = redis_client

    def atomic_counter_increment(self, key, amount=1):
        """Increment a counter within a transaction."""
        pipe = self.r.pipeline()
        pipe.multi()
        pipe.get(key)
        pipe.incrby(key, amount)
        results = pipe.execute()

        old_value = int(results[0]) if results[0] else 0
        new_value = int(results[1])
        return {"old_value": old_value, "new_value": new_value}

    def atomic_batch_set(self, items, ttl=3600):
        """Atomically set multiple keys with expiry."""
        pipe = self.r.pipeline(transaction=True)
        pipe.multi()

        for key, value in items:
            pipe.setex(key, ttl, value)

        results = pipe.execute()
        return {"items_set": len(results), "success": all(results)}

    def atomic_read_modify_write(self, key, modifier_fn):
        """Atomically read, modify, and write a value."""
        pipe = self.r.pipeline()
        while True:
            try:
                pipe.watch(key)
                current = pipe.get(key)
                new_value = modifier_fn(int(current) if current else 0)

                pipe.multi()
                pipe.set(key, new_value)
                pipe.execute()
                return {"old": int(current) if current else 0, "new": new_value}
            except redis.WatchError:
                continue

tx = CacheTransaction(r)

result = tx.atomic_counter_increment("tx:visitors")
print(f"Visitor count: {result['old']} -> {result['new']}")

items = [("tx:user:1", "Alice"), ("tx:user:2", "Bob"), ("tx:user:3", "Charlie")]
result = tx.atomic_batch_set(items, ttl=300)
print(f"Batch set: {result['items_set']} items, success: {result['success']}")

result = tx.atomic_read_modify_write("tx:score", lambda x: x + 10)
print(f"Score: {result['old']} -> {result['new']}")

import redis
import threading
import time

r = redis.Redis(decode_responses=True)

class OptimisticCacheLock:
    def __init__(self, redis_client):
        self.r = redis_client

    def transfer_points(self, from_key, to_key, amount):
        """Atomically transfer points between two users."""
        pipe = self.r.pipeline()
        try:
            pipe.watch(from_key, to_key)

            from_balance = int(pipe.get(from_key) or 0)
            to_balance = int(pipe.get(to_key) or 0)

            if from_balance < amount:
                pipe.unwatch()
                return {"success": False, "reason": "insufficient_balance"}

            pipe.multi()
            pipe.decrby(from_key, amount)
            pipe.incrby(to_key, amount)
            pipe.execute()

            return {
                "success": True,
                "from": from_balance - amount,
                "to": to_balance + amount,
                "transferred": amount,
            }
        except redis.WatchError:
            return {"success": False, "reason": "conflict_retry"}

    def concurrent_transfer_demo(self):
        """Demonstrate optimistic locking with concurrent transfers."""
        self.r.set("points:alice", 100)
        self.r.set("points:bob", 50)

        results = []
        def transfer():
            result = self.transfer_points("points:alice", "points:bob", 30)
            results.append(result)

        threads = [threading.Thread(target=transfer) for _ in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        return results

lock = OptimisticCacheLock(r)

result = lock.transfer_points("points:alice", "points:bob", 30)
print(f"Transfer: {result}")

results = lock.concurrent_transfer_demo()
success_count = sum(1 for r in results if r.get("success"))
conflict_count = sum(1 for r in results if r.get("reason") == "conflict_retry")
print(f"Concurrent transfers: {success_count} succeeded, {conflict_count} conflicts")

import redis

r = redis.Redis(decode_responses=True)

class LuaCacheAtomic:
    def __init__(self, redis_client):
        self.r = redis_client

    def atomic_rate_limiter(self, key, max_requests, window_seconds):
        """Atomically check and increment a rate limit counter."""
        script = """
        local key = KEYS[1]
        local max = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local now = redis.call('TIME')[1]

        redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
        local count = redis.call('ZCARD', key)

        if count >= max then
            return {0, count}
        end

        redis.call('ZADD', key, now, now .. ':' .. math.random())
        redis.call('EXPIRE', key, window)
        return {1, count + 1}
        """
        return self.r.eval(script, 1, key, max_requests, window_seconds)

    def atomic_cache_fill(self, key, fetch_fn, ttl=3600):
        """Atomically check cache, fetch if missing, store result."""
        script = """
        local key = KEYS[1]
        local ttl = tonumber(ARGV[1])
        local cached = redis.call('GET', key)
        if cached then
            return {1, cached}
        end
        return {0, 'miss'}
        """
        result = self.r.eval(script, 1, key, ttl)
        if result[0] == 1:
            return {"source": "cache", "data": result[1]}

        data = fetch_fn()
        self.r.setex(key, ttl, data)
        return {"source": "fetched", "data": data}

rate_limiter = LuaCacheAtomic(r)

for i in range(5):
    allowed, count = rate_limiter.atomic_rate_limiter(
        "ratelimit:api:user_42", 3, 60
    )
    status = "ALLOWED" if allowed == 1 else "DENIED"
    print(f"Request {i+1}: {status} (count: {count})")