cache_manager¶

SyncStrategy ¶

Bases: str, Enum

Strategy for syncing cache to disk.

CachedSettings ¶

Bases: BaseModel, Generic[T]

Manages a cache of values with a configurable history limit.

When a new value is added and the cache is full, the oldest value is removed.

Attributes:

Name	Type	Description
`values`	`list[T]`	List of cached values, newest first
`max_history`	`int`	Maximum number of items to retain in cache

Example

cache = CachedSettingsstr cache.add("first") cache.add("second") cache.get_all() ['second', 'first']

add ¶

add(value: T) -> None

Add a new value to the cache.

If the value already exists, it's moved to the front. If the cache is full, the oldest value is removed.

Parameters:

Name	Type	Description	Default
`value`	`T`	The value to add to the cache	required

Source code in src/clabe/cache_manager.py

def add(self, value: T) -> None:
    """
    Add a new value to the cache.

    If the value already exists, it's moved to the front.
    If the cache is full, the oldest value is removed.

    Args:
        value: The value to add to the cache
    """
    if value in self.values:
        self.values.remove(value)
    self.values.insert(0, value)

    if len(self.values) > self.max_history:
        self.values = self.values[: self.max_history]

get_all ¶

get_all() -> list[T]

Get all cached values.

Returns:

Type	Description
`list[T]`	List of all cached values, newest first

Source code in src/clabe/cache_manager.py

def get_all(self) -> list[T]:
    """
    Get all cached values.

    Returns:
        List of all cached values, newest first
    """
    return self.values.copy()

get_latest ¶

get_latest() -> T | None

Get the most recently added value.

Returns:

Type	Description
`T \| None`	The latest value, or None if cache is empty

Source code in src/clabe/cache_manager.py

def get_latest(self) -> T | None:
    """
    Get the most recently added value.

    Returns:
        The latest value, or None if cache is empty
    """
    return self.values[0] if self.values else None

clear ¶

clear() -> None

Clear all values from the cache.

Source code in src/clabe/cache_manager.py

def clear(self) -> None:
    """Clear all values from the cache."""
    self.values = []

CacheData ¶

Bases: BaseModel

Pydantic model for cache serialization.

CacheManager ¶

CacheManager(
    cache_path: Path | str | None = None,
    sync_strategy: SyncStrategy = AUTO,
)

Thread-safe singleton cache manager with multiple named caches.

Uses Pydantic for proper serialization/deserialization with automatic disk synchronization support. All operations are thread-safe.

Example

Get singleton instance with manual sync (default)¶

manager = CacheManager.get_instance() manager.add_to_cache("subjects", "mouse_001") manager.save() # Explicitly save

Get instance with auto sync - saves after every change¶

manager = CacheManager.get_instance(sync_strategy=SyncStrategy.AUTO) manager.add_to_cache("subjects", "mouse_002") # Automatically saved

Custom path¶

manager = CacheManager.get_instance(cache_path="custom/cache.json")

Initialize a CacheManager instance.

Parameters:

Name	Type	Description	Default
`cache_path`	`Path \| str \| None`	Path to cache file. If None, uses default location.	`None`
`sync_strategy`	`SyncStrategy`	Strategy for syncing to disk (MANUAL or AUTO)	`AUTO`

Source code in src/clabe/cache_manager.py

def __init__(
    self,
    cache_path: Path | str | None = None,
    sync_strategy: SyncStrategy = SyncStrategy.AUTO,
) -> None:
    """
    Initialize a CacheManager instance.

    Args:
        cache_path: Path to cache file. If None, uses default location.
        sync_strategy: Strategy for syncing to disk (MANUAL or AUTO)
    """
    self.caches: dict[str, CachedSettings[Any]] = {}
    self.sync_strategy: SyncStrategy = sync_strategy
    self.cache_path: Path = Path(cache_path) if cache_path else Path(TMP_DIR) / ".cache_manager.json"
    self._instance_lock: threading.RLock = threading.RLock()

get_instance `classmethod` ¶

get_instance(
    cache_path: Path | str | None = None,
    sync_strategy: SyncStrategy = AUTO,
    reset: bool = False,
) -> CacheManager

Get the singleton instance of CacheManager (thread-safe).

Parameters:

Name	Type	Description	Default
`cache_path`	`Path \| str \| None`	Path to cache file. If None, uses default location.	`None`
`sync_strategy`	`SyncStrategy`	Strategy for syncing to disk (MANUAL or AUTO)	`AUTO`
`reset`	`bool`	If True, reset the singleton and create a new instance	`False`

Returns:

Type	Description
`CacheManager`	The singleton CacheManager instance

Source code in src/clabe/cache_manager.py

@classmethod
def get_instance(
    cls,
    cache_path: Path | str | None = None,
    sync_strategy: SyncStrategy = SyncStrategy.AUTO,
    reset: bool = False,
) -> "CacheManager":
    """
    Get the singleton instance of CacheManager (thread-safe).

    Args:
        cache_path: Path to cache file. If None, uses default location.
        sync_strategy: Strategy for syncing to disk (MANUAL or AUTO)
        reset: If True, reset the singleton and create a new instance

    Returns:
        The singleton CacheManager instance
    """
    with cls._lock:
        if reset or cls._instance is None:
            if cache_path is None:
                cache_path = Path(TMP_DIR) / ".cache_manager.json"
            else:
                cache_path = Path(cache_path)

            instance = cls(cache_path=cache_path, sync_strategy=sync_strategy)

            if cache_path.exists():
                try:
                    with cache_path.open("r", encoding="utf-8") as f:
                        cache_data = CacheData.model_validate_json(f.read())
                        instance.caches = cache_data.caches
                except Exception as e:
                    logger.warning("Cache file %s is corrupted: %s. Creating new instance.", cache_path, e)

            cls._instance = instance

        return cls._instance

register_cache ¶

register_cache(
    name: str, max_history: int = _DEFAULT_MAX_HISTORY
) -> None

Register a new cache with a specific history limit (thread-safe).

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique name for the cache	required
`max_history`	`int`	Maximum number of items to retain	`_DEFAULT_MAX_HISTORY`

Source code in src/clabe/cache_manager.py

def register_cache(self, name: str, max_history: int = _DEFAULT_MAX_HISTORY) -> None:
    """
    Register a new cache with a specific history limit (thread-safe).

    Args:
        name: Unique name for the cache
        max_history: Maximum number of items to retain
    """
    with self._instance_lock:
        if name not in self.caches:
            self.caches[name] = CachedSettings(max_history=max_history)
            self._auto_save()

add_to_cache ¶

add_to_cache(name: str, value: Any) -> None

Add a value to a named cache (thread-safe).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the cache	required
`value`	`Any`	Value to add	required

Raises:

Type	Description
`KeyError`	If cache name is not registered

Source code in src/clabe/cache_manager.py

def add_to_cache(self, name: str, value: Any) -> None:
    """
    Add a value to a named cache (thread-safe).

    Args:
        name: Name of the cache
        value: Value to add

    Raises:
        KeyError: If cache name is not registered
    """
    with self._instance_lock:
        if name not in self.caches:
            self.caches[name] = CachedSettings(max_history=_DEFAULT_MAX_HISTORY)

        cache = self.caches[name]

        # we remove it first to avoid duplicates
        if value in cache.values:
            cache.values.remove(value)
        # but add it to the front
        cache.values.insert(0, value)

        if len(cache.values) > cache.max_history:
            cache.values = cache.values[: cache.max_history]

        self._auto_save()

get_cache ¶

get_cache(name: str) -> list[Any]

Get all values from a named cache (thread-safe).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the cache	required

Returns:

Type	Description
`list[Any]`	List of cached values, newest first

Raises:

Type	Description
`KeyError`	If cache name is not registered

Source code in src/clabe/cache_manager.py

def get_cache(self, name: str) -> list[Any]:
    """
    Get all values from a named cache (thread-safe).

    Args:
        name: Name of the cache

    Returns:
        List of cached values, newest first

    Raises:
        KeyError: If cache name is not registered
    """
    with self._instance_lock:
        if name not in self.caches:
            raise KeyError(f"Cache '{name}' not registered.")
        return self.caches[name].values.copy()

try_get_cache ¶

try_get_cache(name: str) -> Any | None

Attempt to get all values from a named cache, returning None if not found.

Source code in src/clabe/cache_manager.py

def try_get_cache(self, name: str) -> Any | None:
    """Attempt to get all values from a named cache, returning None if not found."""
    try:
        return self.get_cache(name)
    except KeyError:
        return None

get_latest ¶

get_latest(name: str) -> Any | None

Get the most recent value from a named cache (thread-safe).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the cache	required

Returns:

Type	Description
`Any \| None`	The latest value, or None if cache is empty

Raises:

Type	Description
`KeyError`	If cache name is not registered

Source code in src/clabe/cache_manager.py

def get_latest(self, name: str) -> Any | None:
    """
    Get the most recent value from a named cache (thread-safe).

    Args:
        name: Name of the cache

    Returns:
        The latest value, or None if cache is empty

    Raises:
        KeyError: If cache name is not registered
    """
    with self._instance_lock:
        values = self.get_cache(name)
        return values[0] if values else None

clear_cache ¶

clear_cache(name: str) -> None

Clear all values from a named cache (thread-safe).

Parameters:

Name	Type	Description	Default
`name`	`str`	Name of the cache	required

Raises:

Type	Description
`KeyError`	If cache name is not registered

Source code in src/clabe/cache_manager.py

def clear_cache(self, name: str) -> None:
    """
    Clear all values from a named cache (thread-safe).

    Args:
        name: Name of the cache

    Raises:
        KeyError: If cache name is not registered
    """
    with self._instance_lock:
        if name not in self.caches:
            raise KeyError(f"Cache '{name}' not registered.")
        self.caches[name].values = []
        self._auto_save()

clear_all_caches ¶

clear_all_caches() -> None

Clear all caches (thread-safe).

Source code in src/clabe/cache_manager.py

def clear_all_caches(self) -> None:
    """Clear all caches (thread-safe)."""
    with self._instance_lock:
        self.caches = {}
        self._auto_save()

save ¶

save() -> None

Save all caches to disk using Pydantic serialization (thread-safe).

This method is called automatically if sync_strategy is AUTO, or can be called manually for MANUAL strategy.

Source code in src/clabe/cache_manager.py

def save(self) -> None:
    """
    Save all caches to disk using Pydantic serialization (thread-safe).

    This method is called automatically if sync_strategy is AUTO,
    or can be called manually for MANUAL strategy.
    """
    with self._instance_lock:
        self._save_unlocked()

cache_manager¶

SyncStrategy ¶

CachedSettings ¶

add ¶

get_all ¶

get_latest ¶

clear ¶

CacheData ¶

CacheManager ¶

Get singleton instance with manual sync (default)¶

Get instance with auto sync - saves after every change¶

Custom path¶

get_instance classmethod ¶

register_cache ¶

add_to_cache ¶

get_cache ¶

try_get_cache ¶

get_latest ¶

clear_cache ¶

clear_all_caches ¶

save ¶

get_instance `classmethod` ¶