Hybrid Caching: Combining Redis, Memcached, and Local Caches

import redis
import time
import json

r = redis.Redis(decode_responses=True)

class HybridCacheRouter:
    def __init__(self, redis_client):
        self.r = redis_client
        self.local_cache = {}

    def classify_data(self, data_type):
        """Classify data to determine the best cache tier."""
        classifications = {
            "session": {
                "tier": "memcached",
                "ttl": 1800,
                "reason": "Simple key-value, high throughput, no persistence needed"
            },
            "user_profile": {
                "tier": "redis",
                "ttl": 3600,
                "reason": "Complex data structure (hashes), needs TTL, persistence"
            },
            "auth_token": {
                "tier": "local",
                "ttl": 300,
                "reason": "Ultra-fast, per-instance, small dataset, frequently accessed"
            },
            "leaderboard": {
                "tier": "redis",
                "ttl": 60,
                "reason": "Sorted set operations, atomic updates"
            },
            "static_config": {
                "tier": "memcached",
                "ttl": 86400,
                "reason": "Simple key-value, rarely changes, high read rate"
            },
            "page_cache": {
                "tier": "redis",
                "ttl": 300,
                "reason": "Needs TTL, cache invalidation, Pub/Sub"
            },
        }
        return classifications.get(data_type, {"tier": "redis", "ttl": 3600})

    def store(self, key, data_type, data):
        """Store data in the appropriate cache tier."""
        classification = self.classify_data(data_type)
        tier = classification["tier"]
        ttl = classification["ttl"]

        if tier == "local":
            self.local_cache[key] = {"data": data, "expires": time.time() + ttl}
            return {"tier": "local", "ttl": ttl}

        serialized = json.dumps(data)
        self.r.setex(key, ttl, serialized)
        return {"tier": tier, "ttl": ttl, "note": classification["reason"]}

    def get(self, key, data_type):
        """Get data from the appropriate cache tier."""
        classification = self.classify_data(data_type)
        tier = classification["tier"]

        if tier == "local":
            entry = self.local_cache.get(key)
            if entry and time.time() < entry["expires"]:
                return {"tier": "local", "data": entry["data"]}
            return None

        cached = self.r.get(key)
        if cached:
            return {"tier": "redis", "data": json.loads(cached)}
        return None

    def report(self):
        """Report cache tier usage."""
        return {
            "local_cache_size": len(self.local_cache),
            "redis_keys": self.r.dbsize(),
        }

router = HybridCacheRouter(r)

test_cases = [
    ("auth:token:42", "auth_token", {"user": 42, "token": "abc"}),
    ("session:abc", "session", {"user": 42, "cart": []}),
    ("user:42:profile", "user_profile", {"name": "Alice", "preferences": {}}),
    ("leaderboard:top", "leaderboard", [{"user": 1, "score": 100}]),
]

for key, data_type, data in test_cases:
    result = router.store(key, data_type, data)
    print(f"Stored {data_type:20s} in {result['tier']:10s} (TTL: {result['ttl']}s)")

for key, data_type, _ in test_cases:
    result = router.get(key, data_type)
    if result:
        print(f"Retrieved {data_type:20s} from {result['tier']}")
    else:
        print(f"Retrieved {data_type:20s}: miss")

print(f"\nCache report: {router.report()}")

import redis
import json

r = redis.Redis(decode_responses=True)

class CostOptimizedCache:
    def __init__(self, redis_client):
        self.r = redis_client
        self.local_cache = {}
        self.tier_costs = {
            "local": {"cost_per_gb_month": 0, "latency": "1μs", "max_size_mb": 100},
            "memcached": {"cost_per_gb_month": 0.15, "latency": "200μs", "max_size_gb": 64},
            "redis": {"cost_per_gb_month": 0.50, "latency": "1ms", "max_size_gb": 512},
        }

    def estimate_cost(self, data_type, size_bytes, access_rate):
        """Estimate monthly cost for different cache tiers."""
        size_gb = size_bytes / (1024 ** 3)
        monthly_requests = access_rate * 3600 * 24 * 30

        costs = {}
        for tier, info in self.tier_costs.items():
            storage_cost = info["cost_per_gb_month"] * size_gb
            if tier == "local":
                if size_gb * 1024 > info["max_size_mb"]:
                    costs[tier] = {"viable": False, "reason": "Exceeds local capacity"}
                    continue
            costs[tier] = {
                "viable": True,
                "monthly_cost": round(storage_cost, 2),
                "latency": info["latency"],
            }
        return costs

    def recommend_tier(self, data_type, size_bytes, access_rate):
        """Recommend the most cost-effective cache tier."""
        costs = self.estimate_cost(data_type, size_bytes, access_rate)
        viable = {t: c for t, c in costs.items() if isinstance(c, dict) and c.get("viable")}

        if not viable:
            return {"tier": "no_cache", "reason": "None viable, use direct DB"}

        cheapest = min(viable, key=lambda t: viable[t]["monthly_cost"])
        return {
            "recommended_tier": cheapest,
            "costs": viable,
            "latency": viable[cheapest]["latency"],
            "monthly_cost": viable[cheapest]["monthly_cost"],
        }

optimizer = CostOptimizedCache(r)

scenarios = [
    ("Auth tokens", 500, 500000),
    ("User profiles", 50_000_000_000, 10000),
    ("Session data", 5_000_000_000, 50000),
    ("Product catalog", 200_000_000_000, 5000),
]

for name, size, rate in scenarios:
    rec = optimizer.recommend_tier(name, size, rate)
    print(f"{name:20s} size={size/1e9:.1f}GB rate={rate:,}/s -> "
          f"{rec['recommended_tier']:10s} ${rec.get('monthly_cost', 0):.2f}/mo "
          f"({rec.get('latency', 'N/A')})")