patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
import sqlite3
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Sequence
|
|
|
|
from .migrate import MigrationError, run_migrations
|
|
from .migrations import MIGRATIONS
|
|
|
|
__all__ = ["Database", "MIGRATIONS", "MigrationError"]
|
|
|
|
|
|
class Database:
|
|
def __init__(self, path, migrations: Sequence[str] | None = None):
|
|
if str(path) != ":memory:":
|
|
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
|
self._conn = sqlite3.connect(str(path), check_same_thread=False)
|
|
self._conn.row_factory = sqlite3.Row
|
|
self._conn.execute("PRAGMA journal_mode=WAL")
|
|
self._conn.execute("PRAGMA foreign_keys=ON")
|
|
self._lock = threading.Lock()
|
|
self._migrations = MIGRATIONS if migrations is None else list(migrations)
|
|
try:
|
|
self._migrate()
|
|
except MigrationError:
|
|
self._conn.close()
|
|
raise
|
|
|
|
def _migrate(self) -> None:
|
|
with self._lock:
|
|
run_migrations(self._conn, self._migrations)
|
|
|
|
def execute(self, sql: str, params: tuple = ()) -> int:
|
|
with self._lock:
|
|
cursor = self._conn.execute(sql, params)
|
|
self._conn.commit()
|
|
return cursor.rowcount
|
|
|
|
def insert(self, sql: str, params: tuple = ()) -> int:
|
|
with self._lock:
|
|
cursor = self._conn.execute(sql, params)
|
|
self._conn.commit()
|
|
return cursor.lastrowid
|
|
|
|
def execute_many(self, statements: list[tuple[str, tuple]]) -> None:
|
|
with self._lock:
|
|
try:
|
|
for sql, params in statements:
|
|
self._conn.execute(sql, params)
|
|
self._conn.commit()
|
|
except Exception:
|
|
self._conn.rollback()
|
|
raise
|
|
|
|
def query(self, sql: str, params: tuple = ()) -> list[sqlite3.Row]:
|
|
with self._lock:
|
|
return self._conn.execute(sql, params).fetchall()
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._conn.close()
|