TTL Tuning: Setting Optimal Cache Expiration Times

import redis
from datetime import datetime, timedelta

r = redis.Redis(decode_responses=True)

class TTLPolicy:
    def __init__(self):
        self.policies = {
            "static": {"ttl": 86400, "description": "24 hours"},
            "semi_static": {"ttl": 3600, "description": "1 hour"},
            "volatile": {"ttl": 300, "description": "5 minutes"},
            "session": {"ttl": 1800, "description": "30 minutes"},
            "rate_limit": {"ttl": 60, "description": "1 minute"},
        }

    def cache_with_policy(self, key, data, category, extra_ttl=0):
        """Cache data with the TTL policy for the given category."""
        policy = self.policies.get(category)
        if not policy:
            raise ValueError(f"Unknown category: {category}")

        ttl = policy["ttl"] + extra_ttl
        if isinstance(data, str):
            r.setex(key, ttl, data)
        else:
            import json
            r.setex(key, ttl, json.dumps(data))

        return {
            "key": key,
            "category": category,
            "ttl_seconds": ttl,
            "expires_at": (datetime.now() + timedelta(seconds=ttl)).isoformat()
        }

    def get_with_category(self, key):
        """Get cached value and report remaining TTL."""
        remaining = r.ttl(key)
        if remaining == -2:
            return None
        value = r.get(key)
        return {"value": value, "remaining_ttl": remaining}

policy = TTLPolicy()

result = policy.cache_with_policy("config:site_name", "DodaTech", "static")
print(f"Static config: TTL={result['ttl_seconds']}s")

result = policy.cache_with_policy("user:42:feed", ["post1", "post2"], "semi_static")
print(f"User feed: TTL={result['ttl_seconds']}s")

result = policy.cache_with_policy("session:abc123", {"user": 42}, "session")
print(f"Session: TTL={result['ttl_seconds']}s")

fetched = policy.get_with_category("config:site_name")
print(f"Retrieved: {fetched['value']}, remaining TTL: {fetched['remaining_ttl']}s")

import redis
import json
from datetime import datetime, timedelta

r = redis.Redis(decode_responses=True)

class FreshnessBasedTTL:
    def __init__(self):
        self.slas = {
            "stock_price": {"max_age_seconds": 5, "refresh_cost": "high"},
            "weather": {"max_age_seconds": 300, "refresh_cost": "medium"},
            "news_headlines": {"max_age_seconds": 600, "refresh_cost": "low"},
            "user_profile": {"max_age_seconds": 3600, "refresh_cost": "medium"},
        }

    def compute_ttl(self, data_type, access_frequency=1.0):
        """Compute optimal TTL based on SLA and access frequency."""
        sla = self.slas[data_type]
        max_age = sla["max_age_seconds"]

        if sla["refresh_cost"] == "high":
            ttl_multiplier = 0.8
        elif sla["refresh_cost"] == "medium":
            ttl_multiplier = 0.9
        else:
            ttl_multiplier = 1.0

        ttl = int(max_age * ttl_multiplier)
        print(f"Data: {data_type}, Max age: {max_age}s, TTL: {ttl}s")
        return ttl

    def cache_fresh(self, key, data, data_type):
        """Cache data with freshness-based TTL."""
        ttl = self.compute_ttl(data_type)
        r.setex(key, ttl, json.dumps(data))
        return {"key": key, "ttl": ttl, "cached_at": datetime.now().isoformat()}

freshness = FreshnessBasedTTL()

items = [
    ("price:AAPL", {"symbol": "AAPL", "price": 245.30}, "stock_price"),
    ("weather:NYC", {"city": "NYC", "temp": 72}, "weather"),
    ("news:top", ["headline1", "headline2"], "news_headlines"),
]

for key, data, data_type in items:
    result = freshness.cache_fresh(key, data, data_type)
    print(f"Cached {key} with TTL {result['ttl']}s")

import redis
import json
import time

r = redis.Redis(decode_responses=True)

class AdaptiveTTLCache:
    def __init__(self, min_ttl=60, max_ttl=86400):
        self.min_ttl = min_ttl
        self.max_ttl = max_ttl
        self.access_counts = {}

    def get(self, key, fetch_fn):
        """Get with adaptive TTL increase on repeated access."""
        value = r.get(key)
        if value is not None:
            self.access_counts[key] = self.access_counts.get(key, 0) + 1
            remaining = r.ttl(key)
            if remaining < 60 and self.access_counts[key] > 5:
                new_ttl = min(remaining * 2, self.max_ttl)
                r.expire(key, new_ttl)
                print(f"Extended TTL for {key} to {new_ttl}s (accessed {self.access_counts[key]}x)")
            return json.loads(value)

        self.access_counts[key] = 0
        data = fetch_fn()
        ttl = self.min_ttl
        r.setex(key, ttl, json.dumps(data))
        print(f"Fresh fetch for {key}, TTL={ttl}s")
        return data

cache = AdaptiveTTLCache()

def fetch_user(user_id):
    print(f"  DB: fetching user {user_id}")
    return {"id": user_id, "name": f"User {user_id}"}

import json as _json
for i in range(8):
    user = cache.get("user:adapt", lambda: fetch_user(42))
    print(f"  Request {i+1}: {user['name']}")
    time.sleep(0.1)