src/data/PersistentTTLCache.py

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

logger = logging.getLogger(__name__)


class PersistentTTLCache:
    """
    Minimal persistent TTL cache backed by SQLite.

    Use case: cache expensive or rate-limited API responses across app restarts.
    """

    def __init__(self, db_path: str = "src/data/.persistent_cache.sqlite3"):
        self.db_path = db_path
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self._init_db()

    def _connect(self) -> sqlite3.Connection:
        # autocommit for small operations
        return sqlite3.connect(self.db_path, isolation_level=None)

    def _init_db(self) -> None:
        with self._connect() as conn:
            conn.execute(
                """
                CREATE TABLE IF NOT EXISTS cache (
                    namespace TEXT NOT NULL,
                    key TEXT NOT NULL,
                    value_json TEXT NOT NULL,
                    expires_at INTEGER NOT NULL,
                    created_at INTEGER NOT NULL,
                    PRIMARY KEY (namespace, key)
                )
                """
            )
            conn.execute(
                "CREATE INDEX IF NOT EXISTS idx_cache_expires ON cache(namespace, expires_at)"
            )

    def get(self, namespace: str, key: str) -> Optional[Dict[str, Any]]:
        now = int(time.time())
        with self._connect() as conn:
            row = conn.execute(
                "SELECT value_json, expires_at FROM cache WHERE namespace=? AND key=?",
                (namespace, key),
            ).fetchone()
            if not row:
                return None
            value_json, expires_at = row
            if int(expires_at) <= now:
                try:
                    conn.execute(
                        "DELETE FROM cache WHERE namespace=? AND key=?",
                        (namespace, key),
                    )
                except Exception:
                    logger.exception("Failed deleting expired cache row")
                return None
            try:
                return json.loads(value_json)
            except Exception:
                logger.exception("Failed decoding cache JSON")
                return None

    def set(
        self, namespace: str, key: str, value: Dict[str, Any], ttl_seconds: int
    ) -> None:
        now = int(time.time())
        expires_at = now + int(ttl_seconds)
        value_json = json.dumps(value, ensure_ascii=True, separators=(",", ":"))
        with self._connect() as conn:
            conn.execute(
                """
                INSERT INTO cache(namespace, key, value_json, expires_at, created_at)
                VALUES(?,?,?,?,?)
                ON CONFLICT(namespace, key) DO UPDATE SET
                    value_json=excluded.value_json,
                    expires_at=excluded.expires_at,
                    created_at=excluded.created_at
                """,
                (namespace, key, value_json, expires_at, now),
            )