cache¶

def is_expired(self, ttl_seconds: float) -> bool:
    """Check if entry has expired based on TTL."""
    if ttl_seconds <= 0:
        return False  # No TTL, never expires
    return time.time() - self.created_at > ttl_seconds

def touch(self) -> None:
    """Update last access time and increment counter."""
    self.last_accessed = time.time()
    self.access_count += 1

def get(self, key: str) -> Optional[T]:
    """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 entry.is_expired(self.ttl_seconds):
            del self._cache[key]
            self.misses += 1
            return None

        entry.touch()
        self.hits += 1
        return entry.value

def set(self, key: str, value: T, size_bytes: int = 0) -> None:
    """Set value in cache.

    Args:
        key: Cache key
        value: Value to cache
        size_bytes: Optional size estimate for memory tracking
    """
    with self._lock:
        # Evict if at capacity
        if len(self._cache) >= self.max_size and key not in self._cache:
            self._evict_lru()

        self._cache[key] = CacheEntry(
            value=value,
            size_bytes=size_bytes,
        )

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

    Args:
        key: Cache key

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

def clear(self) -> int:
    """Clear all entries.

    Returns:
        Number of entries cleared
    """
    with self._lock:
        count = len(self._cache)
        self._cache.clear()
        self.hits = 0
        self.misses = 0
        return count

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

    Returns:
        Dictionary with cache statistics
    """
    with self._lock:
        total_size = sum(e.size_bytes for e in self._cache.values())
        return {
            "name": self.name,
            "size": self.size,
            "max_size": self.max_size,
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate": self.hit_rate,
            "total_bytes": total_size,
            "ttl_seconds": self.ttl_seconds,
        }

def get(self, path: Path) -> Optional[str]:
    """Get cached file content if still valid.

    Args:
        path: Path to file

    Returns:
        File content or None if not cached/stale
    """
    key = str(path.resolve())
    entry = self._cache.get(key)

    if entry is None:
        return None

    mtime, content = entry

    # Check if file was modified
    try:
        current_mtime = path.stat().st_mtime
        if current_mtime > mtime:
            self._cache.delete(key)
            return None
    except OSError:
        self._cache.delete(key)
        return None

    return content

def set(self, path: Path, content: str) -> bool:
    """Cache file content.

    Args:
        path: Path to file
        content: File content

    Returns:
        True if cached, False if file too large
    """
    if len(content) > self.max_file_size:
        return False

    key = str(path.resolve())
    try:
        mtime = path.stat().st_mtime
    except OSError:
        mtime = time.time()

    self._cache.set(key, (mtime, content), size_bytes=len(content))
    return True

def invalidate(self, path: Path) -> bool:
    """Invalidate cached content for a file.

    Args:
        path: Path to file

    Returns:
        True if entry was removed, False if not cached
    """
    return self._cache.delete(str(path.resolve()))

def clear(self) -> int:
    """Clear all cached content.

    Returns:
        Number of entries cleared
    """
    return self._cache.clear()

def stats(self) -> Dict[str, Any]:
    """Get cache statistics."""
    return self._cache.stats()

def get(self, text: str, model: str = "default") -> Optional[list[float]]:
    """Get cached embedding.

    Args:
        text: Text that was embedded
        model: Model used for embedding

    Returns:
        Cached embedding vector or None
    """
    key = cache_key(text[:500], model)  # Truncate for key
    return self._cache.get(key)

def set(
    self,
    text: str,
    embedding: list[float],
    model: str = "default",
) -> None:
    """Cache an embedding.

    Args:
        text: Text that was embedded
        embedding: Embedding vector
        model: Model used for embedding
    """
    key = cache_key(text[:500], model)
    self._cache.set(key, embedding, size_bytes=len(embedding) * 8)

def clear(self) -> int:
    """Clear all cached embeddings.

    Returns:
        Number of entries cleared
    """
    return self._cache.clear()

def stats(self) -> Dict[str, Any]:
    """Get cache statistics."""
    return self._cache.stats()

def get(
    self,
    file_path: Path,
    prompt_hash: str,
    file_mtime: float,
    algorithm: str = "balanced",
) -> Optional[Dict[str, Any]]:
    """Get cached ranking score.

    Args:
        file_path: Path to file
        prompt_hash: Hash of the prompt
        file_mtime: File modification time
        algorithm: Ranking algorithm used

    Returns:
        Cached score data or None
    """
    key = cache_key(str(file_path), prompt_hash, algorithm)
    entry = self._cache.get(key)

    if entry is None:
        return None

    # Check if file was modified since caching
    if entry.get("mtime", 0) < file_mtime:
        self._cache.delete(key)
        return None

    return entry

def set(
    self,
    file_path: Path,
    prompt_hash: str,
    file_mtime: float,
    score: float,
    factors: Dict[str, float],
    algorithm: str = "balanced",
) -> None:
    """Cache a ranking score.

    Args:
        file_path: Path to file
        prompt_hash: Hash of the prompt
        file_mtime: File modification time
        score: Computed relevance score
        factors: Individual ranking factors
        algorithm: Ranking algorithm used
    """
    key = cache_key(str(file_path), prompt_hash, algorithm)
    self._cache.set(
        key,
        {
            "score": score,
            "factors": factors,
            "mtime": file_mtime,
            "algorithm": algorithm,
        },
    )

def clear(self) -> int:
    """Clear all cached scores.

    Returns:
        Number of entries cleared
    """
    return self._cache.clear()

def stats(self) -> Dict[str, Any]:
    """Get cache statistics."""
    return self._cache.stats()