Multi-Tier Caching with Separate Stores

Not all cached data is the same. User session data changes every few seconds. Reference data like country codes changes once a quarter. Product listings change hourly. Pricing tables change weekly.

Using one MemoryStore for everything means a burst of session traffic can evict reference data that rarely changes but is expensive to recompute. The solution: separate stores with different configurations, each tuned to its data's access pattern.

The three-tier pattern

A common pattern is hot, warm, and cold stores.

from inhouse import MemoryStore, inhouse_cache

hot_store = MemoryStore(max_size=512, default_ttl=30, sliding=True)
warm_store = MemoryStore(max_size=2048, default_ttl=300)
cold_store = MemoryStore(max_size=4096, default_ttl=3600)

Tier	max_size	default_ttl	sliding	Use case
Hot	512	30s	Yes	User sessions, rate limit counters
Warm	2048	5min	No	Product listings, search results
Cold	4096	1hr	No	Country codes, tax tables, config

Assigning functions to tiers

Each cached function picks the store that matches its data's volatility.

# Hot tier - user session data, frequently accessed, small footprint
@inhouse_cache(store=hot_store, exclude=("request",))
def load_user_session(request, session_id: str) -> dict | None:
    ...

# Warm tier - catalog data, moderate TTL
@inhouse_cache(store=warm_store, exclude=("db",))
def load_product_list(db, category: str, page: int = 1) -> list[dict]:
    ...

# Cold tier - reference data, rarely changes, expensive to compute
@inhouse_cache(store=cold_store, exclude=("db",))
def load_tax_rates(db, country_code: str) -> dict:
    ...

A traffic spike on sessions evicts only other session entries. Product listings and tax rates are untouched.

Monitoring each tier independently

With separate stores, you see exactly how each tier is performing.

from prometheus_client import Gauge

def report_tier_stats():
    for name, store in [("hot", hot_store), ("warm", warm_store), ("cold", cold_store)]:
        s = store.stats()
        total = s["hits"] + s["misses"]
        ratio = s["hits"] / total if total else 0
        print(f"{name}: {s['size']} entries, {ratio:.1%} hit ratio, {s['evictions']} evictions")

        if s["evictions"] > s["sets"] * 0.1:
            print(f"  Warning: {name} store eviction rate exceeds 10%")

If the hot store shows high evictions, bump max_size. If the cold store shows a low hit ratio, increase default_ttl.

Dynamic rebalancing

Adjust store sizes at runtime based on observed traffic. No restart needed.

def rebalance_stores():
    hot_stats = hot_store.stats()
    cold_stats = cold_store.stats()

    hot_eviction_rate = hot_stats["evictions"] / max(hot_stats["sets"], 1)
    cold_hit_ratio = cold_stats["hits"] / max(cold_stats["hits"] + cold_stats["misses"], 1)

    if hot_eviction_rate > 0.1:
        # Grow hot store, shrink cold store
        new_hot_size = hot_store.size * 2
        print(f"Growing hot store to {new_hot_size} (eviction rate {hot_eviction_rate:.1%})")

    if cold_hit_ratio < 0.5 and cold_stats["size"] < cold_store.max_size * 0.3:
        # Cold store is underutilized - reduce its max_size
        print(f"Cold store is cold ({cold_hit_ratio:.1%} hit ratio) - consider reducing capacity")

Run rebalance_stores() periodically or from a management command.

Shared TTL but isolated capacity

You can configure two stores with the same TTL but different max_size to isolate traffic.

# Same TTL, different capacity
anon_store = MemoryStore(max_size=256, default_ttl=60)
auth_store = MemoryStore(max_size=2048, default_ttl=60)

@inhouse_cache(store=anon_store)
def load_public_page(path: str) -> str:
    ...

@inhouse_cache(store=auth_store)
def load_dashboard(user_id: int) -> dict:
    ...

A viral article driving anonymous traffic evicts only public page cache entries. Authenticated user dashboards are unaffected.

Copy-on-read for sensitive tiers

For the hot store with frequently mutated data, enable copy_on_read to prevent callers from corrupting cached values.

hot_store = MemoryStore(
    max_size=512,
    default_ttl=30,
    sliding=True,
    copy_on_read=True,  # prevent mutation of cached session data
)

Each get() returns a deep copy. Callers can modify the returned dict without affecting the cache.

The recipe in practice

from inhouse import MemoryStore, inhouse_cache
from inhouse.fastapi import create_lifespan
from inhouse.sqlite import query_store, rows_to_dicts

# Three tiers
hot_store = MemoryStore(max_size=512, default_ttl=30, sliding=True)
warm_store = query_store(max_size=2048, default_ttl=300)  # copy_on_read=True
cold_store = MemoryStore(max_size=4096, default_ttl=3600)

@inhouse_cache(store=hot_store, exclude=("db",))
def current_session(db, session_token: str) -> dict | None:
    ...

@inhouse_cache(store=warm_store, exclude=("db",))
def team_dashboard(db, team_id: int, period: str) -> dict:
    ...

@inhouse_cache(store=cold_store, exclude=("db",))
def compliance_rules(db, jurisdiction: str) -> list[dict]:
    ...

# Health check
@app.get("/system/cache-tiers")
async def cache_tiers():
    return {
        "hot": hot_store.stats(),
        "warm": warm_store.stats(),
        "cold": cold_store.stats(),
    }

Each tier is sized and timed for its data. The hot store is small and fast with sliding TTL for sessions. The warm store has copy-on-read for SQLite results. The cold store is large with a long TTL for reference data. Hit ratios and eviction rates are visible per-tier.