home-docker/home-apps/demo-service/_db.py

import os
import json
import sqlite3
import threading
from pathlib import Path
from typing import Optional

DB_PATH = Path(os.environ.get("DEMO_DB_PATH", "/data/demos/.meta/sites.db"))
_lock = threading.Lock()
_conn: Optional[sqlite3.Connection] = None

VALID_VISIBILITY = frozenset({"disabled", "internal", "public"})


def get_conn() -> sqlite3.Connection:
    global _conn
    if _conn is None:
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        _conn = sqlite3.connect(str(DB_PATH), check_same_thread=False)
        _conn.row_factory = sqlite3.Row
        _init_schema(_conn)
    return _conn


def _init_schema(conn: sqlite3.Connection):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS sites (
            name TEXT PRIMARY KEY,
            enabled INTEGER NOT NULL DEFAULT 1,
            published INTEGER NOT NULL DEFAULT 0,
            visibility TEXT NOT NULL DEFAULT 'internal',
            description TEXT NOT NULL DEFAULT '',
            created_at TEXT NOT NULL,
            updated_at TEXT NOT NULL
        )
    """)
    # Migration: add visibility column to existing tables (from v1 schema)
    try:
        conn.execute("ALTER TABLE sites ADD COLUMN visibility TEXT NOT NULL DEFAULT 'internal'")
    except sqlite3.OperationalError:
        pass  # column already exists
    # Migrate old enabled/published data to visibility
    conn.execute("""
        UPDATE sites SET visibility = 'disabled' WHERE enabled = 0 AND visibility = 'internal'
    """)
    conn.execute("""
        UPDATE sites SET visibility = 'public' WHERE enabled = 1 AND published = 1 AND visibility = 'internal'
    """)
    # Future-proof: ensure no invalid visibility values
    conn.execute("""
        UPDATE sites SET visibility = 'internal' WHERE visibility NOT IN ('disabled', 'internal', 'public')
    """)
    conn.commit()


# ─── Converters ───────────────────────────


def _visibility_from_old(enabled: bool, published: bool) -> str:
    """Convert old enabled+published to new visibility."""
    if not enabled:
        return "disabled"
    if published:
        return "public"
    return "internal"


def _row_to_dict(r) -> dict:
    """Convert a SQLite Row to dict with both old and new fields."""
    return {
        "name": r["name"],
        "enabled": r["visibility"] != "disabled",
        "published": r["visibility"] == "public",
        "visibility": r["visibility"],
        "description": r["description"],
        "created_at": r["created_at"],
        "updated_at": r["updated_at"],
    }


# ─── Migration ────────────────────────────


def migrate_from_json():
    """One-time migration from old sites.json to SQLite DB."""
    json_path = DB_PATH.parent / "sites.json"
    if not json_path.exists():
        return False
    conn = get_conn()
    cur = conn.execute("SELECT COUNT(*) FROM sites")
    if cur.fetchone()[0] > 0:
        # Already migrated
        return False
    try:
        data = json.loads(json_path.read_text())
        for name, info in data.get("sites", {}).items():
            published = info.get("published", True)
            enabled = info.get("enabled", True)
            visibility = _visibility_from_old(enabled, published)
            conn.execute(
                """INSERT OR REPLACE INTO sites
                   (name, enabled, published, visibility, description, created_at, updated_at)
                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (name, 1 if enabled else 0, 1 if published else 0, visibility,
                 info.get("description", ""),
                 info.get("created_at", ""),
                 info.get("updated_at", "")),
            )
        conn.commit()
        json_path.rename(json_path.with_suffix(".json.migrated"))
        return True
    except Exception as e:
        print(f"Migration failed: {e}")
        return False


# ─── CRUD ────────────────────────────────


def load_all() -> dict:
    """Return sites as {name: info_dict} for backward compat."""
    conn = get_conn()
    rows = conn.execute(
        "SELECT name, visibility, description, created_at, updated_at FROM sites"
    ).fetchall()
    sites = {}
    for r in rows:
        sites[r["name"]] = _row_to_dict(r)
    return {"sites": sites}


def get(name: str) -> Optional[dict]:
    conn = get_conn()
    r = conn.execute(
        "SELECT name, visibility, description, created_at, updated_at FROM sites WHERE name = ?",
        (name,),
    ).fetchone()
    if not r:
        return None
    return _row_to_dict(r)


def insert(name: str, visibility: str, description: str, now: str):
    if visibility not in VALID_VISIBILITY:
        raise ValueError(f"Invalid visibility: {visibility}")
    enabled = 1 if visibility != "disabled" else 0
    published = 1 if visibility == "public" else 0
    with _lock:
        conn = get_conn()
        conn.execute(
            """INSERT INTO sites (name, enabled, published, visibility, description, created_at, updated_at)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (name, enabled, published, visibility, description, now, now),
        )
        conn.commit()


def update(name: str, **kwargs):
    if not kwargs:
        return
    # Handle visibility specially — sync enabled/published
    if "visibility" in kwargs:
        v = kwargs["visibility"]
        if v not in VALID_VISIBILITY:
            raise ValueError(f"Invalid visibility: {v}")
        kwargs["enabled"] = 1 if v != "disabled" else 0
        kwargs["published"] = 1 if v == "public" else 0
    cols = []
    vals = []
    for k, v in kwargs.items():
        cols.append(f"{k} = ?")
        vals.append(v)
    vals.append(name)
    with _lock:
        conn = get_conn()
        conn.execute(f"UPDATE sites SET {', '.join(cols)} WHERE name = ?", vals)
        conn.commit()


def delete(name: str):
    with _lock:
        conn = get_conn()
        conn.execute("DELETE FROM sites WHERE name = ?", (name,))
        conn.commit()


# Run migration on import
migrate_from_json()