simple_db.py

import sqlite3
import threading
from typing import Any, Optional


class SimpleDB:
    """Thread-safe simple sqlite wrapper for small frequent reads/writes.

    Uses a single connection per process with a lock. Designed for low-latency local access.
    """
    def __init__(self, path: str = ':memory:'):
        self._path = path
        self._conn = sqlite3.connect(self._path, check_same_thread=False)
        self._conn.execute('PRAGMA journal_mode=WAL')
        self._conn.execute('PRAGMA synchronous=NORMAL')
        self._lock = threading.Lock()
        self._ensure_tables()

    def _ensure_tables(self):
        with self._lock:
            cur = self._conn.cursor()
            cur.execute('''
            CREATE TABLE IF NOT EXISTS kv (
                k TEXT PRIMARY KEY,
                v TEXT
            )''')
            self._conn.commit()

    def get(self, key: str) -> Optional[str]:
        with self._lock:
            cur = self._conn.cursor()
            cur.execute('SELECT v FROM kv WHERE k=?', (key,))
            r = cur.fetchone()
            return r[0] if r else None

    def set(self, key: str, value: Any) -> None:
        with self._lock:
            cur = self._conn.cursor()
            cur.execute('REPLACE INTO kv(k,v) VALUES(?,?)', (key, str(value)))
            self._conn.commit()

    def incr(self, key: str, delta: int = 1) -> int:
        with self._lock:
            cur = self._conn.cursor()
            cur.execute('SELECT v FROM kv WHERE k=?', (key,))
            r = cur.fetchone()
            val = int(r[0]) if r and r[0].isdigit() else 0
            val += delta
            cur.execute('REPLACE INTO kv(k,v) VALUES(?,?)', (key, str(val)))
            self._conn.commit()
            return val