142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""DB Explorer — read-only view of all SQLite databases in data/."""
|
|
|
|
import asyncio
|
|
import logging
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
|
|
from config import settings
|
|
from dashboard.templating import templates
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/db-explorer", tags=["db-explorer"])
|
|
|
|
DATA_DIR = Path(settings.repo_root) / "data"
|
|
|
|
# Cap rows per table to avoid blowing up the response
|
|
MAX_ROWS = 200
|
|
|
|
|
|
def _discover_databases() -> list[dict]:
|
|
"""Return metadata for every .db file in data/."""
|
|
if not DATA_DIR.exists():
|
|
return []
|
|
dbs = []
|
|
for path in sorted(DATA_DIR.glob("*.db")):
|
|
try:
|
|
size = path.stat().st_size
|
|
except OSError:
|
|
size = 0
|
|
dbs.append({"name": path.stem, "path": str(path), "size_kb": round(size / 1024, 1)})
|
|
return dbs
|
|
|
|
|
|
def _query_database(db_path: str) -> dict:
|
|
"""Open a database read-only and return all tables with their rows."""
|
|
result = {"tables": {}, "error": None}
|
|
try:
|
|
with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
tables = conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
|
|
).fetchall()
|
|
for (table_name,) in tables:
|
|
try:
|
|
rows = conn.execute(
|
|
f"SELECT * FROM [{table_name}] LIMIT {MAX_ROWS}" # noqa: S608
|
|
).fetchall()
|
|
columns = (
|
|
[
|
|
desc[0]
|
|
for desc in conn.execute(
|
|
f"SELECT * FROM [{table_name}] LIMIT 0"
|
|
).description
|
|
]
|
|
if rows
|
|
else []
|
|
) # noqa: S608
|
|
if not columns and rows:
|
|
columns = list(rows[0].keys())
|
|
elif not columns:
|
|
# Get columns even for empty tables
|
|
cursor = conn.execute(f"PRAGMA table_info([{table_name}])") # noqa: S608
|
|
columns = [r[1] for r in cursor.fetchall()]
|
|
count = conn.execute(f"SELECT COUNT(*) FROM [{table_name}]").fetchone()[0] # noqa: S608
|
|
result["tables"][table_name] = {
|
|
"columns": columns,
|
|
"rows": [dict(r) for r in rows],
|
|
"total_count": count,
|
|
"truncated": count > MAX_ROWS,
|
|
}
|
|
except Exception as exc:
|
|
logger.exception("Failed to query table %s", table_name)
|
|
result["tables"][table_name] = {
|
|
"error": str(exc),
|
|
"columns": [],
|
|
"rows": [],
|
|
"total_count": 0,
|
|
"truncated": False,
|
|
}
|
|
except Exception as exc:
|
|
logger.exception("Failed to query database %s", db_path)
|
|
result["error"] = str(exc)
|
|
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# HTML page
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("", response_class=HTMLResponse)
|
|
async def db_explorer_page(request: Request, db: str = ""):
|
|
"""Render the DB explorer page. If ?db=name, show that database's tables."""
|
|
databases = await asyncio.to_thread(_discover_databases)
|
|
selected = None
|
|
tables_data = {}
|
|
|
|
if db:
|
|
db_path = DATA_DIR / f"{db}.db"
|
|
if db_path.exists():
|
|
selected = db
|
|
result = await asyncio.to_thread(_query_database, str(db_path))
|
|
tables_data = result.get("tables", {})
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"db_explorer.html",
|
|
{
|
|
"databases": databases,
|
|
"selected_db": selected,
|
|
"tables_data": tables_data,
|
|
},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# JSON API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/api/databases", response_class=JSONResponse)
|
|
async def api_list_databases():
|
|
"""List all SQLite databases."""
|
|
return JSONResponse(await asyncio.to_thread(_discover_databases))
|
|
|
|
|
|
@router.get("/api/databases/{db_name}", response_class=JSONResponse)
|
|
async def api_query_database(db_name: str):
|
|
"""Return all tables and rows for a specific database."""
|
|
db_path = DATA_DIR / f"{db_name}.db"
|
|
if not db_path.exists():
|
|
return JSONResponse({"error": f"Database '{db_name}' not found"}, status_code=404)
|
|
result = await asyncio.to_thread(_query_database, str(db_path))
|
|
return JSONResponse(result)
|