App Cache: Patterns & Examples

Practical patterns for self.cache, the async key-value store available on every App instance. Each pattern addresses a specific problem with a complete, runnable example. The Overview covers setup, TTL, 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 = await 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
        await self.cache.set(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".

await 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 = await self.cache.get("motion_count", default=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
        await self.cache.set("motion_count", self.motion_count)
        self.logger.info("Total motion events: %s", self.motion_count)

await self.cache.get("motion_count", default=0) returns the stored value, or 0 when no entry exists yet. 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, name="weather_poll"
        )

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

        # Check cache first
        entry = await self.cache.get(cache_key)
        if entry is not None:
            cached_time, data = entry
            # 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)
        await self.cache.set(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 a ttl parameter in seconds. The entry is removed the moment it's read past that window:

await self.cache.set("weather_data", payload, ttl=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}"

        entry = await self.cache.get(cache_key)
        if entry is not None:
            timestamp, value = entry

            # 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}"
        await self.cache.set(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 = await self.cache.get(
            "energy_stats",
            default=EnergyStats(0.0, 0.0, self.now()),
        )

        await self.scheduler.run_hourly(self.update_stats, name="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
        await self.cache.set("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 is a SQLite read. For values read many times per second, loading into an instance variable at initialization avoids repeated reads:

from hassette import App, AppConfig


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

        # 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 the cache at shutdown
        await self.cache.set("config", self.config_data)

on_initialize reads from the cache 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. 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 call is await self.cache.set("key", value), not an assignment to a local dict.
• Missing await. self.cache.set(...) without await returns a coroutine object and never runs — no error, no write, no log message. Every data method on self.cache is a coroutine.
• 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}/{app_key}/{index}/cache/ — the Hassette process must own the directory.
• Stored value is not picklable. Unpicklable objects raise PicklingError at write time.

Cache Grows Large

The cache has no size limit — entries persist until deleted, expired, or explicitly cleared. Set a ttl on entries that don't need to live forever, and call await self.cache.clear() to delete all entries and reclaim disk space (it runs PRAGMA incremental_vacuum after deleting, so the file actually shrinks). await self.cache.invalidate(*keys) deletes a specific set of keys in one call.

The cache directory at {data_dir}/{app_key}/{index}/cache/ should contain a cache.db file after the first successful write.

See Also

• App Cache Overview. How it works, TTL configuration, lifecycle.
• Global Settings. data_dir reference.