orchestrata_ultimate.core.lru_cache

Reference › orchestrata_ultimate › core › lru_cache

A shared cache utility in core, written so that any package can cache without depending on another package (REQ-010-011). LRUCache stores string-keyed entries in an OrderedDict guarded by one threading.Lock, which gives constant-time lookup and eviction. Each entry is a CacheEntry holding the value, its creation time, its time-to-live and a hit counter. get returns None and counts a miss when the key is missing or its monotonic-clock age exceeds the TTL (the expired entry is deleted); a hit moves the entry to the most-recent end. The constructor takes max_size (default 1000) and default_ttl in seconds (default 300). The cache also keeps overall hit and miss counters.

At a glance

Class diagram

Dependency diagram

API

Classes

LRUCache

Line 16

LRU cache with TTL support (REQ-010-011).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/lru_cache.py

# mypy: ignore-errors
"""Thread-safe LRU cache with TTL support.

Shared utility for any package needing caching without cross-package dependencies.
"""

import collections
import threading
import time
from dataclasses import dataclass
from typing import Any, TypeVar

T = TypeVar("T")


class LRUCache[T]:
    """LRU cache with TTL support (REQ-010-011).

    Thread-safe implementation using OrderedDict for O(1) access and eviction.
    """

    @dataclass
    class CacheEntry:
        """Cache entry with metadata."""

        value: Any
        created_at: float
        ttl_seconds: float
        hits: int = 0

    def __init__(self, max_size: int = 1000, default_ttl: float = 300.0) -> None:
        """Initialize cache.

        Args:
            max_size: Maximum cache entries
            default_ttl: Default TTL in seconds
        """
        self._max_size = max_size
        self._default_ttl = default_ttl
        self._cache: collections.OrderedDict[str, LRUCache.CacheEntry] = (
            collections.OrderedDict()
        )
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> T | None:
        """Get value from cache.

        Args:
            key: Cache key

        Returns:
            Cached value or None if not found/expired
        """
        with self._lock:
            entry = self._cache.get(key)
            if entry is None:
                self._misses += 1
                return None

            if time.monotonic() - entry.created_at > entry.ttl_seconds:
                del self._cache[key]
                self._misses += 1
                return None

            self._cache.move_to_end(key)
            entry.hits += 1
            self._hits += 1
            return entry.value

    def put(self, key: str, value: T, ttl: float | None = None) -> None:
        """Put value in cache.

        Args:
            key: Cache key
            value: Value to cache
            ttl: Optional TTL override
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]

            while len(self._cache) >= self._max_size:
                self._cache.popitem(last=False)

            self._cache[key] = LRUCache.CacheEntry(
                value=value,
                created_at=time.monotonic(),
                ttl_seconds=ttl or self._default_ttl,
            )

    def invalidate(self, key: str) -> bool:
        """Invalidate cache entry.

        Args:
            key: Cache key

        Returns:
            True if entry was found and removed
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                return True
            return False

    def clear(self) -> None:
        """Clear entire cache."""
        with self._lock:
            self._cache.clear()

    def get_stats(self) -> dict[str, Any]:
        """Get cache statistics."""
        with self._lock:
            total = self._hits + self._misses
            hit_rate = (self._hits / total * 100) if total > 0 else 0
            return {
                "size": len(self._cache),
                "max_size": self._max_size,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate_percent": hit_rate,
            }