cache_functions.py - main - Pavz/Project_case3

cache_functions.py
126 строк6 KB
pavel
add all
05 дек 2025, 22:41
05 дек 2025, 22:41c7779b7
import sqlite3
import json
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'utils')) # добавляем папку
import logging
from typing import Optional, Any, Union
from config import DB_NAME, TABLE_NAME


# Настройка логгера
logger = logging.getLogger(__name__)

class CacheManager:
    """
    Класс для управления персистентным кэшем, основанным на SQLite3.
    Хранит результаты дорогостоящих вычислений (0/1, векторы).
    """

    def __init__(self, db_name: str = DB_NAME, table_name: str = TABLE_NAME):
        self.db_name = db_name
        self.table_name = table_name
        self.conn: Optional[sqlite3.Connection] = None # Без Optional анализатор типов будет считать, что conn всегда имеет тип sqlite3.Connection, и не предупредит, если попытатсья вызвать метод на None:


    def connect(self):
        """
        Устанавливает соединение с базой данных и создает таблицу кэша если не существует, IF NOT EXISTS ... 
        """
        try:
            self.conn = sqlite3.connect(self.db_name)
            self.conn.row_factory = sqlite3.Row  # Позволяет получать результаты как словари (по умолчанию возвращает строки как обычные кортежи)
            # работает только если имена столбцов указаны явно (например, SELECT name, price, а не SELECT *  )
            cursor = self.conn.cursor()
            
            # Создание таблицы:
            # key: Уникальный SHA256 хеш товара
            # value: Результат (строка JSON) - может быть '0', '1', или вектор эмбеддинга
            # timestamp: Для потенциального механизма устаревания (TTL)
            cursor.execute(f'''
                CREATE TABLE IF NOT EXISTS {self.table_name} (
                    key TEXT PRIMARY KEY,
                    value TEXT NOT NULL,
                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            ''')

            self.conn.commit()
            logger.info(f"SQLite кэш подключен: {self.db_name}")

        except sqlite3.Error as e:
            logger.error(f"Ошибка при подключении: {e}")
            self.conn = None

    def close(self):
        """
        Закрывает соединение с базой данных.
        """
        if self.conn:
            self.conn.close()
            logger.info("Соединение с кэшем закрыто.")


    def save_to_cache(self, key, result):
        """
        Сохраняет результат вычисления в кэш. Результат сохраняется в JSON. Предварительно нужно использовать нашу ф-ю connect

        :param key: Уникальный ключ товара (например, SHA256 хеш).
        :param result: Результат (float-ов для эмбеддинга, можно использовать и другое).
        """
        if not self.conn:
            logger.warning("Нет активного соединения с кэшем. Пропуск сохранения.")
            return

        try:
            value_json = json.dumps(result)# преобразуем в json
        except TypeError as e:
            logger.error(f"Ошибка сериализации {key}: {e}")
            return

        sql = f"""
            INSERT OR REPLACE INTO {self.table_name} (key, value)
            VALUES (?, ?)
        """
        try:
            self.conn.execute(sql, (key, value_json)) # при каждом сохранении добвляем ключ и значене (в виде json)
            self.conn.commit()
            logger.debug(f"Ключ {key} успешно сохранен ")
        except sqlite3.Error as e:
            logger.error(f"Ошибка при сохранении {key}: {e}")

    def get_from_cache(self, key: str):
        """
        Извлекает результат из кэша по ключу и десериализует его.

        :param key: Уникальный ключ товара.
        :return: Десериализованный результат (list[float] для эмбеддинга), или None, если ключ не найден.
        """
        if not self.conn:
            logger.warning("Нет активного соединения с кэшем.")
            return None

        sql = f"SELECT value FROM {self.table_name} WHERE key = ?" #  ? обозначают параметры, которые будут подставлены позже.
        try:
            cursor = self.conn.execute(sql, (key,)) # подставляю вместо ? key
            row = cursor.fetchone()

            if row:
                # Десериализация из JSON
                value_json = row['value']
                result = json.loads(value_json) # преобразует json строку в python объект (в нашем случае dict или list)
                logger.debug(f"Ключ {key} найден")
                return result
            
            logger.debug(f"Ключ {key} не найден в кэше")
            return None

        except sqlite3.Error as e:
            logger.error(f"Ошибка при извлечении ключа {key}: {e}")
            return None
        except json.JSONDecodeError as e:
            logger.error(f"Ошибка десериализации  {key}.  {e}")

            # !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!   Удалить поврежденную запись, чтобы избежать повторной ошибки
            self.conn.execute(f"DELETE FROM {self.table_name} WHERE key = ?", (key,))
            self.conn.commit()
            return None