Cache Observability with Store Stats

You added caching and the API got faster. But how much faster? Are you evicting too aggressively? Is the cache actually helping, or are most requests missing?

Before v1.0, the only way to know was to wrap everything in manual counters or guess from response times. MemoryStore.stats() exposes hit, miss, eviction, set, and delete counters directly.

Reading the stats

from inhouse import MemoryStore, inhouse_cache

store = MemoryStore(max_size=1024, default_ttl=60)

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

# After some traffic:
print(store.stats())
# {'hits': 847, 'misses': 153, 'evictions': 12, 'sets': 153, 'deletes': 0, 'size': 141}

The hit ratio is 847 / (847 + 153) = 84.7%. The eviction count tells you how many entries were pushed out by LRU before their TTL expired. If evictions are high, max_size might need tuning.

Tracking stats over time

A simple health endpoint that returns cache metrics:

from fastapi import FastAPI
from inhouse import MemoryStore
from inhouse.fastapi import create_lifespan, fastapi_cache

store = MemoryStore(max_size=1024, default_ttl=60)
app = FastAPI(lifespan=create_lifespan(store))

@app.get("/health/cache")
async def cache_health():
    s = store.stats()
    total = s["hits"] + s["misses"]
    hit_ratio = s["hits"] / total if total > 0 else 0.0
    return {
        "hit_ratio": round(hit_ratio, 3),
        "size": s["size"],
        "evictions": s["evictions"],
        "total_requests": total,
    }

Now you can curl /health/cache and see the cache performance at a glance.

Multiple stores, separate metrics

If you use separate stores for different data, each has its own counters:

hot_store = MemoryStore(max_size=512, default_ttl=30)
cold_store = MemoryStore(max_size=4096, default_ttl=3600)

@inhouse_cache(store=hot_store)
def load_recent_activity(team_id: int) -> list[dict]:
    ...

@inhouse_cache(store=cold_store)
def load_team_settings(team_id: int) -> dict:
    ...

print("Hot store:", hot_store.stats())
print("Cold store:", cold_store.stats())

The hot store should have a high hit ratio (frequently accessed data). The cold store should have very few evictions (large capacity, long TTL). If the hot store shows high evictions, bump its max_size.

Exporting to Prometheus

from prometheus_client import Gauge

cache_hits = Gauge("inhouse_cache_hits", "Total cache hits")
cache_misses = Gauge("inhouse_cache_misses", "Total cache misses")
cache_evictions = Gauge("inhouse_cache_evictions", "LRU eviction count")
cache_size = Gauge("inhouse_cache_size", "Current cache entry count")

def update_cache_metrics():
    s = store.stats()
    cache_hits.set(s["hits"])
    cache_misses.set(s["misses"])
    cache_evictions.set(s["evictions"])
    cache_size.set(s["size"])

Run update_cache_metrics() on a schedule or from a metrics endpoint. Track hit ratio over time to spot regressions after deploys.

Debugging cache effectiveness

You suspect a particular endpoint is not benefiting from caching. Check its store stats before and after.

from inhouse import MemoryStore, inhouse_cache

user_store = MemoryStore(max_size=1024, default_ttl=60)
report_store = MemoryStore(max_size=256, default_ttl=120)

@inhouse_cache(store=user_store)
def get_user(user_id: int) -> dict:
    ...

@inhouse_cache(store=report_store)
def get_report(team_id: int, period: str) -> dict:
    ...

# After a day of traffic:
print("User store:", user_store.stats())
# {'hits': 5230, 'misses': 320, 'evictions': 5, ...}  -> 94% hit ratio

print("Report store:", report_store.stats())
# {'hits': 89, 'misses': 410, 'evictions': 200, ...}  -> 18% hit ratio

The report store has a low hit ratio and high evictions. The max_size=256 is too small for the number of distinct report queries. Bump it to 2048 or increase the TTL so entries live long enough to get reused.

Using stats to tune max_size

The eviction count tells you if your cache is sized correctly.

def recommend_max_size(store: MemoryStore, target_eviction_rate: float = 0.01):
    s = store.stats()
    total = s["sets"]
    if total == 0:
        return None
    eviction_rate = s["evictions"] / total
    if eviction_rate > target_eviction_rate:
        return int(s["size"] * 1.5)  # grow by 50%
    return None

If you see sustained evictions, increase capacity before it impacts hit ratio.

Stats are under the store lock

Counter updates are atomic under the existing store lock. No thread safety concerns. The stats() snapshot is consistent at the moment of the call.

# Thread-safe to call from any handler
snapshot = store.stats()

What stats are NOT for

The counters reset when the process restarts. They are not persisted to disk. They are not shared across workers. They tell you about this process's lifetime, not global cache health. For multi-worker setups, aggregate at the metrics layer.

The recipe in production

from inhouse import MemoryStore, inhouse_cache
from inhouse.fastapi import create_lifespan

store = MemoryStore(max_size=2048, default_ttl=60)

@inhouse_cache(store=store)
def get_catalog(item_id: int) -> dict:
    ...

# Periodic stats logging
def log_cache_stats():
    s = store.stats()
    total = s["hits"] + s["misses"]
    ratio = s["hits"] / total if total else 0.0
    print(f"Cache: {s['size']} entries, {ratio:.1%} hit ratio, {s['evictions']} evictions")

    if s["evictions"] > s["sets"] * 0.1:
        print("Warning: eviction rate exceeds 10%. Consider increasing max_size.")

Run log_cache_stats() on a cron or from a management command to keep an eye on cache health without adding dependencies.