cache/ttl_cache.py - main - Soldier327/LLM_AI

cache/ttl_cache.py
90 строк3 KB
Soldier-327
Initial commit
24 май 2026, 17:24
24 май 2026, 17:24dae6bc3
import logging
import threading
import time
from collections import OrderedDict
from typing import Dict, Any, Optional

logger = logging.getLogger(__name__)


class TTLCache:
    """Кеш с TTL (Time To Live) с использованием словаря"""

    def __init__(self, default_ttl_seconds: int = 3600, max_size: int = 1000):
        self._cache: Dict[str, tuple[Any, float]] = OrderedDict()
        self._default_ttl = default_ttl_seconds
        self._max_size = max_size
        self._lock = threading.Lock()
        logger.info(f"Инициализирован кеш: TTL={default_ttl_seconds}с, max_size={max_size}")

    def _is_expired(self, timestamp: float) -> bool:
        """Проверка, истек ли элемент"""
        return time.time() > timestamp

    def _clean_expired(self):
        """Удаление просроченных элементов"""
        with self._lock:
            expired_keys = [
                key for key, (_, timestamp) in self._cache.items()
                if self._is_expired(timestamp)
            ]
            for key in expired_keys:
                del self._cache[key]
            if expired_keys:
                logger.debug(f"Удалено {len(expired_keys)} просроченных элементов из кеша")

    def get(self, key: str) -> Optional[Any]:
        """Получение значения из кеша"""
        self._clean_expired()

        with self._lock:
            if key not in self._cache:
                return None

            value, timestamp = self._cache[key]
            if self._is_expired(timestamp):
                del self._cache[key]
                return None

            # Перемещаем в конец (для LRU)
            self._cache.move_to_end(key)
            logger.debug(f"Кеш HIT для ключа: {key[:50]}...")
            return value

    def set(self, key: str, value: Any, ttl_seconds: Optional[int] = None):
        """Установка значения в кеш с опциональным TTL"""
        ttl = ttl_seconds or self._default_ttl
        timestamp = time.time() + ttl

        with self._lock:
            # Если кеш переполнен, удаляем самый старый элемент
            if len(self._cache) >= self._max_size:
                oldest_key = next(iter(self._cache))
                del self._cache[oldest_key]
                logger.debug(f"Кеш переполнен, удален элемент: {oldest_key[:50]}...")

            self._cache[key] = (value, timestamp)
            self._cache.move_to_end(key)
            logger.debug(f"Добавлено в кеш: ключ={key[:50]}..., TTL={ttl}с")

    def delete(self, key: str):
        """Удаление элемента из кеша"""
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                logger.debug(f"Удален из кеша: {key[:50]}...")

    def clear(self):
        """Очистка всего кеша"""
        with self._lock:
            self._cache.clear()
            logger.info("Кеш полностью очищен")

    def get_stats(self) -> Dict[str, Any]:
        """Получение статистики кеша"""
        with self._lock:
            return {
                "size": len(self._cache),
                "max_size": self._max_size,
                "default_ttl": self._default_ttl
            }