settled-reach/whatsinagame/static/db/connectors/sqlite_connector.py

#!/usr/bin/env python3
"""
SQLite Connector — mini MCP for ticket management.

Usage:
    python3 sqlite_connector.py init
    python3 sqlite_connector.py query "SELECT * FROM tickets"
    python3 sqlite_connector.py execute "UPDATE tickets SET status='done' WHERE id=1"
    python3 sqlite_connector.py --help
"""

import json
import os
import sqlite3
import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# Paths
# ---------------------------------------------------------------------------

SCRIPT_DIR = Path(__file__).resolve().parent
CONFIG_PATH = SCRIPT_DIR / "config.json"
SCHEMA_PATH = SCRIPT_DIR.parent / "schema.sql"


def load_config():
    """Load config.json."""
    with open(CONFIG_PATH, "r") as f:
        return json.load(f)


def resolve_db_path():
    """Resolve database path from environment or config."""
    env_path = os.environ.get("PROJECT_DB")
    if env_path:
        return Path(env_path).resolve()
    cfg = load_config()
    db_name = cfg.get("db_name", "project.db")
    db_location = cfg.get("db_location", "parent")
    if db_location == "parent":
        return (SCRIPT_DIR / ".." / ".." / ".." / db_name).resolve()
    elif db_location == "local":
        return (SCRIPT_DIR / ".." / ".." / db_name).resolve()
    else:
        return Path(db_location).resolve() / db_name


def get_connection():
    """Return an sqlite3 connection with WAL mode and foreign keys enabled."""
    db_path = resolve_db_path()
    conn = sqlite3.connect(str(db_path))
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA foreign_keys=ON;")
    conn.row_factory = sqlite3.Row
    return conn


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

def cmd_init():
    """Initialise the database from schema.sql."""
    if not SCHEMA_PATH.exists():
        return {"ok": False, "error": f"Schema file not found: {SCHEMA_PATH}"}

    schema_sql = SCHEMA_PATH.read_text()
    conn = get_connection()
    try:
        conn.executescript(schema_sql)
        conn.commit()
        return {"ok": True, "message": f"Database initialised at {resolve_db_path()}"}
    except sqlite3.Error as exc:
        return {"ok": False, "error": str(exc)}
    finally:
        conn.close()


def cmd_query(sql):
    """Run a SELECT query and return results as a JSON array of objects."""
    conn = get_connection()
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
        return {"ok": True, "count": len(rows), "rows": rows}
    except sqlite3.Error as exc:
        return {"ok": False, "error": str(exc)}
    finally:
        conn.close()


def cmd_execute(sql):
    """Run an INSERT/UPDATE/DELETE and return affected row count."""
    conn = get_connection()
    try:
        cursor = conn.execute(sql)
        conn.commit()
        return {
            "ok": True,
            "affected_rows": cursor.rowcount,
            "last_id": cursor.lastrowid,
        }
    except sqlite3.Error as exc:
        return {"ok": False, "error": str(exc)}
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

HELP_TEXT = """\
SQLite Connector

Usage:
  sqlite_connector.py init                  Create/update database from schema.sql
  sqlite_connector.py query "<SQL>"         Run a SELECT and return JSON rows
  sqlite_connector.py execute "<SQL>"       Run INSERT/UPDATE/DELETE, return affected rows
  sqlite_connector.py --help                Show this help message

All output is JSON on stdout. Errors also use JSON with {{"ok": false, "error": "..."}}.

Config:  {config}
Schema:  {schema}
""".format(config=CONFIG_PATH, schema=SCHEMA_PATH)


def main():
    if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"):
        print(HELP_TEXT)
        sys.exit(0)

    cmd = sys.argv[1]

    if cmd == "init":
        result = cmd_init()
    elif cmd == "query":
        if len(sys.argv) < 3:
            result = {"ok": False, "error": "query requires a SQL string argument"}
        else:
            result = cmd_query(sys.argv[2])
    elif cmd == "execute":
        if len(sys.argv) < 3:
            result = {"ok": False, "error": "execute requires a SQL string argument"}
        else:
            result = cmd_execute(sys.argv[2])
    else:
        result = {"ok": False, "error": f"Unknown command: {cmd}. Use --help for usage."}

    print(json.dumps(result, indent=2))
    sys.exit(0 if result.get("ok") else 1)


if __name__ == "__main__":
    main()