spothole/core/single_object_data_cache.py

import logging
import threading

import diskcache

logger = logging.getLogger(__name__)


class SingleObjectDataCache:
    """Cache for status and solar conditions. This uses DiskCache, but unlike the standard DiskCache users like
    activity ref and callsign lookup handlers, status and solar conditions are persisted as a single object. If we
    just load the object from DiskCache and modify it, DiskCache doesn't know that it's been updated and needs
    re-caching, so we provide a store() method that any functions updating the object can call afterwards."""

    def __init__(self, cache_dir, object_if_empty):
        """Initialize a SingleObjectDataCache. Provide the directory to load the cache from and save it to. If the cache
        is empty, the provided object_if_empty parameter will be used to initialise it."""

        self._lock = threading.Lock()
        self._cache = diskcache.Cache(cache_dir)
        # This cache stores a single object, doesn't matter what it's called so "object" will do
        if "object" not in self._cache:
            self._cache.add("object", object_if_empty)
        try:
            self._obj = self._cache.get("object")
        except Exception:
            logger.warning(f"Failed to load cache from {cache_dir}, clearing it.")
            self._cache.clear()
            self._cache.add("object", object_if_empty)
            self._obj = object_if_empty

    def get(self):
        """Get the data object. This can then be manipulated as necessary across multiple threads. Any function
        modifying the object must remember to call store() afterwards."""
        return self._obj

    def store(self):
        """Store the updated object in the cache. Any function modifying the object must remember to call this
        afterwards."""
        with self._lock:
            self._cache.set("object", self._obj)

    def close(self):
        self.store()
        self._cache.close()