Cache System Example

A TTL-based cache system with namespace support.

Full Source

  1"""
  2Cache System Example - sqbooster
  3
  4A TTL-based cache with decorator pattern and function memoization.
  5"""
  6
  7import time
  8import tempfile
  9import os
 10from datetime import datetime, timedelta
 11import sqbooster
 12
 13
 14class CacheManager:
 15    def __init__(self, db):
 16        self.db = db
 17
 18    def set(self, key, value, ttl_seconds=3600):
 19        entry = {
 20            "value": value,
 21            "created_at": datetime.now().isoformat(),
 22            "expires_at": (datetime.now() + timedelta(seconds=ttl_seconds)).isoformat(),
 23        }
 24        self.db.write(f"cache:{key}", entry)
 25
 26    def get(self, key):
 27        entry = self.db.read(f"cache:{key}")
 28        if entry is None:
 29            return None
 30        if datetime.now() > datetime.fromisoformat(entry["expires_at"]):
 31            self.db.delete_key(f"cache:{key}")
 32            return None
 33        return entry["value"]
 34
 35    def delete(self, key):
 36        self.db.delete_key(f"cache:{key}")
 37
 38    def clear(self):
 39        for key in self.db.keys("cache:"):
 40            self.db.delete_key(key)
 41
 42    def stats(self):
 43        keys = self.db.keys("cache:")
 44        total = len(keys)
 45        valid = 0
 46        for key in keys:
 47            entry = self.db.read(key)
 48            if entry and datetime.now() <= datetime.fromisoformat(entry["expires_at"]):
 49                valid += 1
 50        return {"total": total, "valid": valid, "expired": total - valid}
 51
 52
 53def memoize(cache, ttl=60):
 54    def decorator(func):
 55        def wrapper(*args):
 56            cache_key = f"{func.__name__}:{args}"
 57            result = cache.get(cache_key)
 58            if result is not None:
 59                print(f"  [CACHE HIT] {func.__name__}{args}")
 60                return result
 61            print(f"  [CACHE MISS] {func.__name__}{args}")
 62            result = func(*args)
 63            cache.set(cache_key, result, ttl_seconds=ttl)
 64            return result
 65        return wrapper
 66    return decorator
 67
 68
 69def expensive_computation(n):
 70    time.sleep(0.1)
 71    return sum(i * i for i in range(n))
 72
 73
 74def main():
 75    print("=== Cache System Example ===\n")
 76
 77    print("--- SQLite Cache ---")
 78    cache = CacheManager(sqbooster.sqlite())
 79
 80    cache.set("user:1", {"name": "Alice", "role": "admin"}, ttl_seconds=10)
 81    cache.set("user:2", {"name": "Bob", "role": "user"}, ttl_seconds=10)
 82
 83    print(f"  user:1 = {cache.get('user:1')}")
 84    print(f"  user:2 = {cache.get('user:2')}")
 85    print(f"  user:3 = {cache.get('user:3')}")
 86    print(f"  Stats: {cache.stats()}")
 87
 88    print("\n--- Memoized Function ---")
 89    cached_compute = memoize(cache, ttl=30)(expensive_computation)
 90
 91    result1 = cached_compute(100)
 92    print(f"  Result: {result1}")
 93
 94    result2 = cached_compute(100)
 95    print(f"  Result: {result2}")
 96
 97    result3 = cached_compute(200)
 98    print(f"  Result: {result3}")
 99
100    print(f"  Stats: {cache.stats()}")
101
102    print("\n--- JSON Cache ---")
103    tmp = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
104    tmp.close()
105
106    json_cache = CacheManager(sqbooster.json(tmp.name))
107    json_cache.set("theme", {"dark": True, "font_size": 14})
108    print(f"  theme = {json_cache.get('theme')}")
109
110    os.unlink(tmp.name)
111
112    print("\nDone!")
113
114
115if __name__ == "__main__":
116    main()

What It Demonstrates

  • Key-value API for simple storage

  • Backend switching between SQLite and JSON

  • Decorator pattern for function caching

  • Pickle serialization for complex objects