Graph Cache¶

def cache_figure(
    self,
    key: str,
    figure: go.Figure,
    ttl: Optional[int] = None
) -> None:
    """
    Cache a Plotly figure.

    Parameters
    ----------
    key : str
        Cache key
    figure : go.Figure
        Plotly figure to cache
    ttl : Optional[int], default=None
        TTL in seconds
    """
    # Convert figure to JSON dict
    fig_dict = figure.to_dict()

    # Store JSON-serializable dict
    self.set(key, fig_dict, ttl=ttl)

    logger.info(f"Cached figure: {key}")

def get_cached_figure(self, key: str) -> Optional[go.Figure]:
    """
    Retrieve cached Plotly figure.

    Parameters
    ----------
    key : str
        Cache key

    Returns
    -------
    Optional[go.Figure]
        Cached figure or None if not found
    """
    fig_dict = self.get(key)

    if fig_dict is None:
        return None

    # Reconstruct figure from dict
    figure = go.Figure(fig_dict)

    logger.debug(f"Retrieved cached figure: {key}")

    return figure

def generate_figure_key(
    self,
    analysis_id: str,
    filter_values: dict,
    config: Optional[dict] = None
) -> str:
    """
    Generate unique cache key for figure.

    Parameters
    ----------
    analysis_id : str
        Analysis ID (e.g., 'UC1_1')
    filter_values : dict
        Filter values used to generate figure
    config : Optional[dict], default=None
        Additional configuration parameters

    Returns
    -------
    str
        Cache key
    """
    # Create hash from analysis + filters + config
    hash_components = {
        'analysis_id': analysis_id,
        'filters': filter_values,
        'config': config or {}
    }

    hash_string = json.dumps(hash_components, sort_keys=True)
    hash_value = hashlib.md5(
        hash_string.encode()
    ).hexdigest()[:16]

    return f"{analysis_id}_{hash_value}"

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

    Parameters
    ----------
    key : str
        Cache key

    Returns
    -------
    Optional[Any]
        Cached value or None if not found/expired
    """
    # Check if key exists
    if key not in self._cache:
        logger.debug(f"Cache miss: {key}")
        self._misses += 1
        self._emit_operation_metric("get", "miss")
        self._emit_hit_ratio_metric()
        return None

    # Check if expired
    if self._is_expired(key):
        logger.debug(f"Cache expired: {key}")
        self.delete(key)
        self._misses += 1
        self._emit_operation_metric("get", "expired")
        self._emit_hit_ratio_metric()
        return None

    # Move to end (LRU)
    self._cache.move_to_end(key)

    logger.debug(f"Cache hit: {key}")
    self._hits += 1
    self._emit_operation_metric("get", "hit")
    self._emit_hit_ratio_metric()
    return self._cache[key]

def set(
    self,
    key: str,
    value: Any,
    ttl: Optional[int] = None
) -> None:
    """
    Set value in cache.

    Parameters
    ----------
    key : str
        Cache key
    value : Any
        Value to cache
    ttl : Optional[int], default=None
        TTL in seconds (if None, uses default_ttl)
    """
    # Enforce max size (evict oldest if necessary)
    if key not in self._cache and len(self._cache) >= self.max_size:
        self._evict_oldest()

    # Set value
    self._cache[key] = value
    self._cache.move_to_end(key)

    # Set expiry
    ttl = ttl if ttl is not None else self.default_ttl
    if ttl > 0:
        self._expiry[key] = datetime.now() + timedelta(seconds=ttl)
    else:
        self._expiry[key] = None

    logger.debug(
        f"Cache set: {key}",
        extra={'ttl': ttl}
    )
    self._emit_operation_metric("set", "ok")
    CACHE_ENTRY_SIZE_BYTES.labels(cache_type=self._cache_type).observe(
        self._estimate_value_size_bytes(value)
    )
    self._emit_size_metric()

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

    Parameters
    ----------
    key : str
        Cache key

    Returns
    -------
    bool
        True if deleted, False if key not found
    """
    if key in self._cache:
        del self._cache[key]
        del self._expiry[key]
        logger.debug(f"Cache delete: {key}")
        self._emit_operation_metric("delete", "ok")
        self._emit_size_metric()
        return True
    return False

def clear(self) -> None:
    """Clear all cache entries."""
    count = len(self._cache)
    self._cache.clear()
    self._expiry.clear()
    logger.info(f"Cache cleared: {count} entries removed")
    self._emit_operation_metric("clear", "ok")
    self._emit_size_metric()

def exists(self, key: str) -> bool:
    """
    Check if key exists in cache (and not expired).

    Parameters
    ----------
    key : str
        Cache key

    Returns
    -------
    bool
        True if exists and not expired
    """
    if key not in self._cache:
        return False

    if self._is_expired(key):
        self.delete(key)
        return False

    return True

def size(self) -> int:
    """
    Get current cache size.

    Returns
    -------
    int
        Number of entries in cache
    """
    return len(self._cache)

def get_stats(self) -> dict:
    """
    Get cache statistics.

    Returns
    -------
    dict
        Statistics including size, max_size, usage percentage
    """
    return {
        'current_size': len(self._cache),
        'max_size': self.max_size,
        'default_ttl': self.default_ttl,
        'usage_percent': (len(self._cache) / self.max_size) * 100
    }