Ty
0eb2b0f697
- backend/db/schema.sql: all 12 tables from DATA-MODEL.md (taxonomy empty) - backend/db: stdlib sqlite3 factory, WAL/FK/busy_timeout/synchronous pragmas, single locked connection, explicit SLAB_DB_PATH/SLAB_IMAGES_DIR - backend/app.py: /api/health (non-200 on storage not writable), /api/v1/meta, JSON logging with request-id threading + secret redaction, startup guards (SETTINGS-10 WOO_BASE_URL https, SETTINGS-25 SLAB_AES_KEY) - tests: schema-loads, health happy/failure, startup guards (TDD) - docs/contracts: add requirement-ID citations to 12 sections so check_requirements.py passes (pre-existing gate debt) Gates: check_requirements.py + check_coverage.py exit 0; pytest 5 passed
99 lines
2.8 KiB
Python
99 lines
2.8 KiB
Python
"""S0 db factory package: stdlib sqlite3 ONLY. ... (same as before)"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
import threading
|
|
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 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()
|
|
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.close()
|
|
except Exception:
|
|
pass
|
|
_conn = None
|
|
_current_path = None
|