just a simple dict-like sqlite + pickle cache

sqlite_cache.py

import pickle
import sqlite3
from dataclasses import dataclass
from dataclasses import field
from typing import cast


@dataclass
class SQLCache[K, V]:
    """SQLite-backed persistent cache."""

    path: Path
    data_version: int
    max_entries: int = 100_000
    VERSION = 1

    _db: sqlite3.Connection = field(init=False)

    def __post_init__(self) -> None:
        cache_path = self.path / f"cache_{self.VERSION}_{self.data_version}.sqlite"
        cache_path.parent.mkdir(parents=True, exist_ok=True)
        self._db = sqlite3.connect(cache_path)
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute("PRAGMA synchronous=NORMAL")
        self._db.executescript(
            """CREATE TABLE IF NOT EXISTS cache (key BLOB PRIMARY KEY, value BLOB NOT NULL) WITHOUT ROWID;"""
        )

    @staticmethod
    def _dump(obj: object) -> bytes:
        return pickle.dumps(obj, protocol=pickle.HIGHEST_PROTOCOL)

    def get(self, key: K) -> V | None:
        row = self._db.execute("SELECT value FROM cache WHERE key = ?", (self._dump(key),)).fetchone()
        if row is None:
            return None
        return cast(V, pickle.loads(row[0]))  # noqa: S301

    def __setitem__(self, key: K, value: V) -> None:
        self._db.execute(
            """INSERT OR REPLACE INTO cache (key, value) VALUES (?, ?)""", (self._dump(key), self._dump(value))
        )

    def close(self) -> None:
        """honor system: only close it once, don't use after; ProgrammingError if you do."""
        (count,) = self._db.execute("SELECT COUNT(*) FROM cache").fetchone()
        if count > self.max_entries:
            # Drop ~50% arbitrarily  (this will happen maybe once a year probably)
            self._db.execute("""DELETE FROM cache WHERE key IN ( SELECT key FROM cache LIMIT ?)""", (count // 2,))

        self._db.commit()
        self._db.close()