Skip to content

App Cache: Patterns & Examples

Practical patterns for self.cache, the persistent disk-backed key-value store available on every App instance. Each pattern addresses a specific problem with a complete, runnable example. The Overview covers setup and basic usage.

self.now() returns the current time as a ZonedDateTime from the whenever library. It is timezone-aware, picklable, and supports arithmetic — all time-based patterns below use it for timestamp comparisons.

Rate-Limiting Notifications

A leak or alarm sensor can fire repeatedly during a single incident. Storing a timestamp in the cache prevents duplicate notifications from going out during a cooldown window:

from hassette import App, AppConfig, P


class WaterLeakAlertApp(App[AppConfig]):
    async def on_initialize(self):
        await self.bus.on_state_change(
            "binary_sensor.water_leak",
            handler=self.on_leak_detected,
            where=P.StateTo("on"),
            name="water_leak",
        )

    async def on_leak_detected(self, event):
        """Send notification, but not more than once every 4 hours."""
        cache_key = "last_leak_notification"

        # Check when we last sent a notification
        last_sent = self.cache.get(cache_key)

        if last_sent is not None:
            if last_sent > self.now().subtract(hours=4):
                time_since_last = self.now() - last_sent
                self.logger.info("Skipping notification - last sent %s ago", time_since_last)
                return

        # Send the notification
        await self.api.call_service(
            "notify",
            "mobile_app",
            message="Water leak detected!",
            title="Alert",
        )

        # Update cache with current time
        self.cache[cache_key] = self.now()
        self.logger.info("Leak notification sent")

P is hassette.event_handling.predicates — helper functions that filter which events trigger a handler (see Predicates). P.StateTo("on") fires the handler only when the entity transitions to "on".

self.cache.get(cache_key) returns None on the first call, so the notification goes out immediately. The timestamp is written after sending. On subsequent triggers, the handler compares the stored timestamp against the cooldown threshold — last_sent > self.now().subtract(hours=4) is true when the last notification was sent less than 4 hours ago. For per-entity rate limiting, include the entity ID in the key: f"last_notification:{entity_id}".

Persistent Counters

A counter stored only in an instance variable resets to zero whenever Hassette restarts. Loading from the cache at initialization and writing back on every increment makes the counter survive restarts:

from hassette import App, AppConfig, P


class MotionCounterApp(App[AppConfig]):
    async def on_initialize(self):
        # Restore counter from cache, or start at 0
        self.motion_count = self.cache.get("motion_count", 0)
        self.logger.info("Motion count restored: %s", self.motion_count)

        await self.bus.on_state_change(
            "binary_sensor.motion",
            handler=self.on_motion,
            where=P.StateTo("on"),
            name="motion_counter",
        )

    async def on_motion(self, event):
        self.motion_count += 1
        self.cache["motion_count"] = self.motion_count
        self.logger.info("Total motion events: %s", self.motion_count)

self.cache.get("motion_count", 0) returns the stored value or 0 when no entry exists. Each call to on_motion increments the in-memory counter and immediately writes the new value to disk. Restart Hassette and check the logs — "Motion count restored: N" confirms the counter survived.

API Response Caching

External APIs impose rate limits. Storing the response alongside a timestamp lets the app return a cached copy while the data is still fresh:

from hassette import App, AppConfig


class WeatherApp(App[AppConfig]):
    async def on_initialize(self):
        await self.scheduler.run_every(self.update_weather, seconds=60)

    async def get_weather(self, location: str) -> dict:
        cache_key = f"weather:{location}"

        # Check cache first
        if cache_key in self.cache:
            cached_time, data = self.cache[cache_key]
            # Return cached data if less than 30 minutes old
            if cached_time > self.now().subtract(minutes=30):
                self.logger.info("Using cached weather for %s", location)
                return data

        # Fetch fresh data from API
        self.logger.info("Fetching fresh weather for %s", location)
        data = await self.fetch_weather_api(location)
        self.cache[cache_key] = (self.now(), data)
        return data

    async def fetch_weather_api(self, location: str) -> dict:
        # Your external API call here
        return {"temperature": 72}

    async def update_weather(self):
        weather = await self.get_weather("New York")
        await self.api.set_state(
            "sensor.weather_forecast",
            str(weather["temperature"]),
        )

get_weather checks the cache first. The entry holds a tuple of (timestamp, data). When the stored timestamp falls within the 30-minute window, the cached value is returned without a network call. A stale or absent entry triggers a fresh fetch and overwrites the cache entry.

Why the # pyright: ignore comments?

cache.get() returns untyped values — the type checker can't know what was stored under a key. The examples suppress the resulting warnings; production code can do the same, or narrow the value with a cast or an isinstance check after reading.

Expiring Entries

Two approaches exist for expiring cache entries, depending on whether access to the timestamp is needed.

For automatic expiry, self.cache.set() accepts an expire parameter in seconds. The underlying diskcache library removes the entry silently once the timeout elapses:

self.cache.set("weather_data", payload, expire=3600)  # expires in 1 hour

When the timestamp is needed for display or custom staleness logic, storing it explicitly alongside the value works better:

from hassette import App, AppConfig


class DataCacheApp(App[AppConfig]):
    async def get_cached_data(self, key: str, ttl_minutes: int = 60):
        """Get data from cache if not expired, or None if expired or absent."""
        cache_key = f"data:{key}"

        if cache_key in self.cache:
            timestamp, value = self.cache[cache_key]

            # Return cached data if still within TTL
            if timestamp > self.now().subtract(minutes=ttl_minutes):
                return value

        # Data expired or not found
        return None

    async def set_cached_data(self, key: str, value) -> None:
        """Store data alongside a timestamp for TTL tracking."""
        cache_key = f"data:{key}"
        self.cache[cache_key] = (self.now(), value)

get_cached_data compares the stored timestamp against the configured TTL and returns None when the entry is stale. The caller decides whether to re-fetch.

Storing Complex Data

The cache stores any picklable Python object. Dataclasses with typed fields work well for structured app state:

import dataclasses
from dataclasses import dataclass

from hassette import App, AppConfig
from whenever import ZonedDateTime


@dataclass
class EnergyStats:
    total_kwh: float
    peak_usage: float
    last_updated: ZonedDateTime


class EnergyTrackerApp(App[AppConfig]):
    async def on_initialize(self):
        # Load previous stats or create new ones
        self.stats: EnergyStats = self.cache.get(
            "energy_stats",
            EnergyStats(0.0, 0.0, self.now()),
        )

        await self.scheduler.run_hourly(self.update_stats)

    async def update_stats(self):
        current_usage = await self.get_current_usage()

        # Create a new stats object — do not mutate the existing one
        self.stats = dataclasses.replace(
            self.stats,
            total_kwh=self.stats.total_kwh + current_usage,
            peak_usage=max(self.stats.peak_usage, current_usage),
            last_updated=self.now(),
        )

        # Persist to cache
        self.cache["energy_stats"] = self.stats
        self.logger.info("Updated stats: %s", self.stats)

    async def get_current_usage(self) -> float:
        state = await self.api.get_state("sensor.power_usage")
        return float(state.value)

dataclasses.replace() produces a new EnergyStats object rather than modifying the existing one. The cache write only happens after the new object is fully constructed. A runtime error before the write leaves the previous value intact.

Load Once, Write on Shutdown

Cache access involves disk I/O. For values read many times per second, loading into an instance variable at initialization avoids repeated disk reads:

from hassette import App, AppConfig


class OptimizedApp(App[AppConfig]):
    async def on_initialize(self):
        # Load from disk cache once into an instance variable
        self.config_data: dict = self.cache.get("config", {})

        # Use the in-memory copy throughout the app's lifetime
        setting = self.config_data.get("some_setting")
        self.logger.info("Setting: %s", setting)

    async def on_shutdown(self):
        # Persist changes back to disk cache at shutdown
        self.cache["config"] = self.config_data

on_initialize reads from disk once. All access during the run uses the in-memory copy. on_shutdown — the lifecycle hook that mirrors on_initialize, called when Hassette stops cleanly — writes the final state back to disk. diskcache is thread-safe for multi-threaded access; within Hassette's single async event loop, two handlers that touch the same key between await points cannot interleave, but when an await falls between a read and a write, the last write wins — for counters or accumulators, use instance variables as shown in the Persistent Counters pattern.

Troubleshooting

Cache Not Persisting

If values do not survive a restart, check four common causes:

  • Write targets a local variable instead of self.cache. Verify the assignment uses self.cache["key"], not a local dict.
  • Exception during initialization. The app may raise before the write executes. Check hassette log --app <key> for errors.
  • Cache directory lacks write permissions. Check ls -la {data_dir}/{ClassName}/cache/ — the Hassette process must own the directory.
  • Stored value is not picklable. Unpicklable objects raise PicklingError at write time. Enable log_level = "DEBUG" under [hassette.logging] in hassette.toml to see the error.

Cache Size Exceeded

When the cache reaches default_cache_size, diskcache silently evicts the least recently stored entries (oldest writes first — its default policy). A larger default_cache_size in Global Settings raises the ceiling. TTL expiry removes stale entries proactively, and storing large objects externally while caching only their identifiers reduces pressure.

Set log_level = "DEBUG" under [hassette.logging] in hassette.toml to enable cache operation logging. The cache directory at ~/.local/share/hassette/v0/{ClassName}/cache/ (where {ClassName} matches the app's class name, e.g. WaterLeakAlertApp) should contain data files after the first successful write.

See Also