Ty
a0032ddd2d
- Add schema_version table; init_db detects a pre-M3/unknown DB (no version row, or settings table missing the value column) and rebuilds to the clean M3 schema, then stamps version 1 - _has_schema_version checks for a real row (empty table != versioned) - _reset_for_tests checkpoints WAL before close (Windows handle release) - TDD: test_schema_drift seeds the M2 settings shape, asserts rebuild + version Fixes live 500 'no such column: value' on the UAT host (stale M2 slab_db volume). Gates: check_requirements + check_coverage exit 0; pytest 14 passed.
130 lines
4.0 KiB
Python
130 lines
4.0 KiB
Python
"""S0 db factory package: stdlib sqlite3 ONLY. ... (same as before)"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DB_PATH_ENV = "SLAB_DB_PATH"
|
|
IMAGES_DIR_ENV = "SLAB_IMAGES_DIR"
|
|
SCHEMA_FILE = "schema.sql"
|
|
|
|
_lock = threading.Lock()
|
|
_conn: sqlite3.Connection | None = None
|
|
_current_path: str | None = None
|
|
|
|
|
|
def _get_schema_path() -> Path:
|
|
# backend/db/__init__.py -> backend/db/schema.sql
|
|
return Path(__file__).parent / SCHEMA_FILE
|
|
|
|
|
|
def get_connection(db_path: str | None = None) -> sqlite3.Connection:
|
|
global _conn, _current_path
|
|
if db_path is None:
|
|
db_path = os.environ.get(DB_PATH_ENV)
|
|
if not db_path:
|
|
raise RuntimeError(f"{DB_PATH_ENV} env var is required")
|
|
db_path = str(Path(db_path).resolve())
|
|
|
|
with _lock:
|
|
if _conn is None or _current_path != db_path:
|
|
if _conn is not None:
|
|
_conn.close()
|
|
_conn = sqlite3.connect(db_path, check_same_thread=False)
|
|
_conn.row_factory = sqlite3.Row
|
|
_conn.execute("PRAGMA journal_mode=WAL")
|
|
_conn.execute("PRAGMA foreign_keys=ON")
|
|
_conn.execute("PRAGMA busy_timeout=5000")
|
|
_conn.execute("PRAGMA synchronous=NORMAL")
|
|
_current_path = db_path
|
|
return _conn
|
|
|
|
def _has_schema_version(conn: sqlite3.Connection) -> bool:
|
|
try:
|
|
return conn.execute("SELECT 1 FROM schema_version LIMIT 1").fetchone() is not None
|
|
except sqlite3.OperationalError:
|
|
return False
|
|
|
|
def _has_settings_value_column(conn: sqlite3.Connection) -> bool:
|
|
try:
|
|
info = conn.execute("PRAGMA table_info(settings)").fetchall()
|
|
columns = {row[1]: row[2] for row in info}
|
|
return 'value' in columns
|
|
except sqlite3.OperationalError:
|
|
return False
|
|
|
|
|
|
def init_db(db_path: str | None = None) -> sqlite3.Connection:
|
|
conn = get_connection(db_path)
|
|
schema_path = _get_schema_path()
|
|
if not schema_path.exists():
|
|
raise FileNotFoundError(f"Schema not found: {schema_path}")
|
|
with open(schema_path, "r", encoding="utf-8") as f:
|
|
schema = f.read()
|
|
with _lock:
|
|
conn.executescript(schema)
|
|
conn.commit()
|
|
if not _has_schema_version(conn):
|
|
if not _has_settings_value_column(conn):
|
|
logger = logging.getLogger("slab.db")
|
|
logger.info("Detected pre-M3 DB, rebuilding schema")
|
|
conn.executescript("DROP TABLE IF EXISTS settings")
|
|
conn.executescript(schema)
|
|
conn.execute("INSERT OR REPLACE INTO schema_version (version) VALUES (1)")
|
|
else:
|
|
# M3-era tables present but no version stamped; stamp it.
|
|
conn.execute("INSERT OR REPLACE INTO schema_version (version) VALUES (1)")
|
|
conn.commit()
|
|
|
|
return conn
|
|
|
|
|
|
def get_images_dir() -> Path:
|
|
img = os.environ.get(IMAGES_DIR_ENV)
|
|
if not img:
|
|
raise RuntimeError(f"{IMAGES_DIR_ENV} env var is required")
|
|
return Path(img).resolve()
|
|
|
|
|
|
def probe_writable(conn: sqlite3.Connection | None = None) -> bool:
|
|
try:
|
|
c = conn or get_connection()
|
|
c.execute("SELECT 1").fetchone()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def probe_images_writable() -> bool:
|
|
try:
|
|
img_dir = get_images_dir()
|
|
if not img_dir.exists():
|
|
img_dir.mkdir(parents=True, exist_ok=True)
|
|
probe = img_dir / ".health-probe"
|
|
probe.write_text("ok", encoding="utf-8")
|
|
probe.unlink(missing_ok=True)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _reset_for_tests() -> None:
|
|
"""Close global conn so temp dirs can be cleaned on Windows (test only)."""
|
|
global _conn, _current_path
|
|
with _lock:
|
|
if _conn is not None:
|
|
try:
|
|
_conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
|
|
except Exception:
|
|
pass
|
|
try:
|
|
_conn.close()
|
|
except Exception:
|
|
pass
|
|
_conn = None
|
|
_current_path = None
|