cache.py - master - muhxa/into-practice

cache.py
86 строк3 KB
fisson-ui
Добавлены новые сервисы, кэш GigaChat, обновлён интерфейс
06 июл 2026, 14:21
06 июл 2026, 14:21332c313
"""
Простой SQLite-кэш ответов GigaChat.

Идея (см. слайд "Цели кэширования" командного задания):
если клиент запрашивает один и тот же аналитический сервис повторно с тем же
user_id, нет смысла второй раз обращаться к нейросети — токены и время не
бесконечны. Мы кэшируем финальный текстовый ответ GigaChat по паре
(user_id, service), а не «сырые» данные из Loginom — свежие цифры из Loginom
всё равно можно запросить заново в любой момент, они дешёвые (обычный REST),
дорогой ресурс — именно вызов LLM.

Хранилище — файл cache.db рядом с приложением. Для учебного/демо-проекта
этого достаточно; в продакшене вместо SQLite можно подключить Redis с TTL.
"""

import sqlite3
from datetime import datetime, timezone
from pathlib import Path

DB_PATH = Path(__file__).parent / "cache.db"


def _get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS gigachat_cache (
            user_id    INTEGER NOT NULL,
            service    TEXT    NOT NULL,
            response   TEXT    NOT NULL,
            created_at TEXT    NOT NULL,
            PRIMARY KEY (user_id, service)
        )
        """
    )
    return conn


def get_cached(user_id: int, service: str):
    """Вернуть (response, created_at) из кэша или None, если записи нет."""
    conn = _get_conn()
    try:
        cur = conn.execute(
            "SELECT response, created_at FROM gigachat_cache "
            "WHERE user_id = ? AND service = ?",
            (user_id, service),
        )
        return cur.fetchone()
    finally:
        conn.close()


def save_cache(user_id: int, service: str, response: str) -> None:
    """Сохранить (или перезаписать) ответ GigaChat для пары (user_id, service)."""
    conn = _get_conn()
    try:
        conn.execute(
            "INSERT OR REPLACE INTO gigachat_cache "
            "(user_id, service, response, created_at) VALUES (?, ?, ?, ?)",
            (user_id, service, response, datetime.now(timezone.utc).isoformat()),
        )
        conn.commit()
    finally:
        conn.close()


def clear_cache(user_id: int | None = None, service: str | None = None) -> int:
    """
    Удалить записи из кэша (например, кнопка «Обновить рекомендацию» в UI).
    Без аргументов — очищает всё. Возвращает число удалённых строк.
    """
    conn = _get_conn()
    try:
        query = "DELETE FROM gigachat_cache WHERE 1=1"
        params = []
        if user_id is not None:
            query += " AND user_id = ?"
            params.append(user_id)
        if service is not None:
            query += " AND service = ?"
            params.append(service)
        cur = conn.execute(query, params)
        conn.commit()
        return cur.rowcount
    finally:
        conn.close()