forked from Rockachopa/Timmy-time-dashboard
feat: add DB Explorer for read-only SQLite inspection
Adds /db-explorer page and JSON API to browse all 15 SQLite databases in data/. Sidebar lists databases with sizes, clicking one renders all tables as scrollable data tables with row truncation at 200. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
144
src/dashboard/routes/db_explorer.py
Normal file
144
src/dashboard/routes/db_explorer.py
Normal file
@@ -0,0 +1,144 @@
|
||||
"""DB Explorer — read-only view of all SQLite databases in data/."""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import sqlite3
|
||||
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:
|
||||
conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
|
||||
conn.row_factory = sqlite3.Row
|
||||
except Exception as exc:
|
||||
result["error"] = str(exc)
|
||||
return result
|
||||
|
||||
try:
|
||||
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:
|
||||
result["tables"][table_name] = {
|
||||
"error": str(exc),
|
||||
"columns": [],
|
||||
"rows": [],
|
||||
"total_count": 0,
|
||||
"truncated": False,
|
||||
}
|
||||
except Exception as exc:
|
||||
result["error"] = str(exc)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
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)
|
||||
Reference in New Issue
Block a user