cache/base.py - master - YoungFreddy/FinalWork

cache/base.py
69 строк3 KB
Igor
Initial commit: Q&A LLM service
23 июл 2026, 18:02
23 июл 2026, 18:02e5e37db
"""Реализация кэша в памяти с TTL и логированием hit/miss."""

import json
import logging
import time
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)


class SimpleCache:
    """Кэш в памяти с TTL и структурированным логированием."""

    def __init__(self, ttl: int = 600):
        """Инициализирует кэш с указанным TTL (сек). TTL должен быть > 0."""
        if ttl <= 0:
            raise ValueError(f"TTL must be positive, got {ttl}")
        self._cache: Dict[str, Dict[str, Any]] = {}
        self.ttl = ttl

    def _structured_log(self, event: str, **kwargs):
        record = {"event": event, "timestamp": time.time(), **kwargs}
        logger.info(json.dumps(record, ensure_ascii=False))

    def get(self, key: str) -> Optional[str]:
        """Возвращает значение по ключу или None при отсутствии/истечении. Логирует hit/miss."""
        if not key:
            return None
        if key not in self._cache:
            self._structured_log("cache_miss", cache_key=key)
            return None
        entry = self._cache[key]
        if entry["expires_at"] < time.time():
            del self._cache[key]
            self._structured_log("cache_miss_expired", cache_key=key)
            return None
        self._structured_log("cache_hit", cache_key=key)
        return entry["value"]

    def set(self, key: str, value: str) -> None:
        """Сохраняет значение с TTL. Удаляет просроченные записи перед записью."""
        if not key or value is None:
            return
        self.cleanup_expired()
        self._cache[key] = {
            "value": value,
            "expires_at": time.time() + self.ttl,
        }
        self._structured_log("cache_set", cache_key=key, ttl=self.ttl)

    def clear(self) -> None:
        """Очищает весь кэш."""
        size = len(self._cache)
        self._cache.clear()
        self._structured_log("cache_cleared", entries_removed=size)

    def cleanup_expired(self) -> int:
        """Удаляет просроченные записи и возвращает их количество."""
        current_time = time.time()
        expired_keys = [
            key
            for key, entry in self._cache.items()
            if entry["expires_at"] < current_time
        ]
        for key in expired_keys:
            del self._cache[key]
        if expired_keys:
            self._structured_log("cache_cleanup", expired_count=len(expired_keys))
        return len(expired_keys)