Dynamic TTL with Feature Flags
Your cache TTLs are hardcoded. When traffic spikes, you want to lengthen TTLs to reduce database load. When you push a data update, you want to shorten TTLs so fresh data propagates faster. With a callable TTL, you change the TTL at runtime without redeploying.
Callable TTL basics
Instead of a float, pass a callable that returns the TTL in seconds. Inhouse evaluates it on every cache miss.
from inhouse import inhouse_cache, MemoryStore
store = MemoryStore(max_size=1024)
# A mutable setting that controls TTL at runtime
cache_ttl = 60
@inhouse_cache(lambda: cache_ttl, store=store)
def get_user_profile(user_id: int) -> dict:
...
Change cache_ttl and the next cache miss uses the new value. Existing cached entries keep their original TTL until they expire.
Driving TTL from environment variables
Use an environment variable that you can change without restarting.
import os
def ttl_from_env():
return int(os.environ.get("INHOUSE_TTL", "60"))
@inhouse_cache(ttl_from_env, store=store)
def get_product(product_id: int) -> dict:
...
Change the env var and send a SIGHUP (or just wait for the next cache miss - the callable reads os.environ fresh each time). No code change, no deploy.
Feature flag driven TTL
Integrate with a feature flag system to control TTL per-route without touching code.
from featureflags import get_flag # hypothetical FF client
def ttl_for_route(route_name: str):
def resolver():
flag = get_flag(f"cache_ttl_{route_name}")
return flag.value if flag else 60
return resolver
@inhouse_cache(ttl_for_route("user_profile"), store=store)
def get_user_profile(user_id: int) -> dict:
...
@inhouse_cache(ttl_for_route("product_list"), store=store)
def list_products(category: str) -> list[dict]:
...
Toggle cache_ttl_user_profile from 60 to 300 in your feature flag dashboard. The next cache miss for get_user_profile uses 300 seconds. No deploy, no config push.
Gradual rollout of longer TTLs
Use a random factor to gradually increase cache duration, observing system behavior before committing.
import random
def gradual_ttl():
base = 60
if random.random() < 0.1: # 10% of writes use a longer TTL
return base * 5 # 300 seconds
return base
@inhouse_cache(gradual_ttl, store=store)
def get_analytics_report(report_id: str) -> dict:
...
Start with 10% of writes using the longer TTL. Monitor hit ratios and error rates. Ramp up the percentage as confidence grows.
Time-of-day based TTL
Cache longer during peak hours when database load is highest, shorter during off-peak when fresh data matters more.
from datetime import datetime
def ttl_by_time():
hour = datetime.now().hour
if 9 <= hour <= 17: # peak business hours
return 300 # 5 minutes - reduce DB load
else:
return 60 # 1 minute - fresher data when cheap
@inhouse_cache(ttl_by_time, store=store)
def get_dashboard_data(team_id: int) -> dict:
...
Peak hours: longer TTL, less DB load. Off-peak: shorter TTL, fresher data. The transition happens automatically.
Circuit breaker pattern
If the database starts failing, automatically lengthen TTLs to keep the site running on cached data.
import time
db_failure_window = []
def circuit_breaker_ttl():
global db_failure_window
# Count failures in the last 60 seconds
recent = [t for t in db_failure_window if time.time() - t < 60]
db_failure_window = recent
if len(recent) > 5: # more than 5 failures in 60s
return 600 # 10 minutes - keep stale data alive
return 60 # normal TTL
@inhouse_cache(circuit_breaker_ttl, store=store)
def get_critical_config(key: str) -> str:
try:
result = db.fetch_one("SELECT value FROM config WHERE key = ?", (key,))
return result["value"]
except Exception:
db_failure_window.append(time.time())
raise # re-raise - inhouse caches only on success
When the database is healthy, TTL is 60 seconds. When failures accumulate, TTL jumps to 10 minutes - keeping the app running on cached data until the database recovers.
Priority order reminder
TTL is resolved on cache miss:
- Callable
ttl_seconds()result (if callable) - Fixed
ttl_secondsfloat (if provided) store.default_ttl(if set)ValueError
With sliding TTL, the callable is NOT re-evaluated on reads. Sliding extends by the duration stored at write time.