roa2web-service-auto/backend/modules/reports/cache/memory_cache.py

"""
In-memory cache with TTL (L1 cache)
Fast, limited size, lost on restart
"""
import time
import logging
from typing import Any, Optional, Dict
from collections import OrderedDict

logger = logging.getLogger(__name__)


class MemoryCache:
    """
    In-memory LRU cache with TTL support

    Features:
    - LRU eviction when max_size reached
    - Per-entry TTL expiration
    - Thread-safe operations
    - Fast O(1) get/set operations
    """

    def __init__(self, max_size: int = 1000):
        """
        Initialize memory cache

        Args:
            max_size: Maximum number of entries to store
        """
        self.max_size = max_size
        self._cache: OrderedDict[str, Dict[str, Any]] = OrderedDict()
        self._stats = {
            'hits': 0,
            'misses': 0,
            'sets': 0,
            'evictions': 0
        }

    async def get(self, key: str) -> Optional[Any]:
        """
        Get value from cache

        Args:
            key: Cache key

        Returns:
            Cached value or None if not found/expired
        """
        if key not in self._cache:
            self._stats['misses'] += 1
            return None

        entry = self._cache[key]

        # Check TTL expiration
        if entry['expires_at'] < time.time():
            # Expired - remove and return None
            del self._cache[key]
            self._stats['misses'] += 1
            logger.debug(f"Memory cache expired: {key}")
            return None

        # Move to end (LRU - most recently used)
        self._cache.move_to_end(key)

        self._stats['hits'] += 1
        logger.debug(f"Memory cache HIT: {key}")
        return entry['value']

    async def set(self, key: str, value: Any, ttl: int):
        """
        Set value in cache

        Args:
            key: Cache key
            value: Value to cache
            ttl: Time to live in seconds
        """
        expires_at = time.time() + ttl

        # Check if we need to evict (LRU)
        if key not in self._cache and len(self._cache) >= self.max_size:
            # Evict oldest entry (first item in OrderedDict)
            evicted_key = next(iter(self._cache))
            del self._cache[evicted_key]
            self._stats['evictions'] += 1
            logger.debug(f"Memory cache evicted (LRU): {evicted_key}")

        # Store entry
        self._cache[key] = {
            'value': value,
            'expires_at': expires_at,
            'created_at': time.time()
        }

        # Move to end (most recently used)
        self._cache.move_to_end(key)

        self._stats['sets'] += 1
        logger.debug(f"Memory cache SET: {key} (TTL: {ttl}s)")

    async def delete(self, key: str) -> bool:
        """
        Delete entry from cache

        Args:
            key: Cache key

        Returns:
            True if deleted, False if not found
        """
        if key in self._cache:
            del self._cache[key]
            logger.debug(f"Memory cache deleted: {key}")
            return True
        return False

    async def clear(self):
        """Clear all entries from cache"""
        count = len(self._cache)
        self._cache.clear()
        logger.info(f"Memory cache cleared: {count} entries removed")

    async def clear_by_pattern(self, pattern: str):
        """
        Clear entries matching pattern (simple prefix match)

        Args:
            pattern: Key prefix to match (e.g., "dashboard_summary:123")
        """
        keys_to_delete = [key for key in self._cache.keys() if key.startswith(pattern)]
        for key in keys_to_delete:
            del self._cache[key]

        logger.info(f"Memory cache cleared by pattern '{pattern}': {len(keys_to_delete)} entries")

    async def cleanup_expired(self):
        """Remove all expired entries"""
        now = time.time()
        expired_keys = [
            key for key, entry in self._cache.items()
            if entry['expires_at'] < now
        ]

        for key in expired_keys:
            del self._cache[key]

        if expired_keys:
            logger.info(f"Memory cache cleanup: {len(expired_keys)} expired entries removed")

    def get_stats(self) -> Dict[str, Any]:
        """
        Get cache statistics

        Returns:
            Dictionary with stats (hits, misses, size, etc.)
        """
        total_requests = self._stats['hits'] + self._stats['misses']
        hit_rate = (self._stats['hits'] / total_requests * 100) if total_requests > 0 else 0

        return {
            'size': len(self._cache),
            'max_size': self.max_size,
            'hits': self._stats['hits'],
            'misses': self._stats['misses'],
            'sets': self._stats['sets'],
            'evictions': self._stats['evictions'],
            'hit_rate': hit_rate,
            'total_requests': total_requests
        }

    def reset_stats(self):
        """Reset statistics counters"""
        self._stats = {
            'hits': 0,
            'misses': 0,
            'sets': 0,
            'evictions': 0
        }