From a4c9b60079102f2e8d90cf1d22593b1fc1b99d19 Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 15:42:37 +0000 Subject: [PATCH] =?UTF-8?q?feat(ui):=20Data=20Sources=20UI=20=E2=80=94=20e?= =?UTF-8?q?nterprise=20browser=20for=20all=205=20source=20databases?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New "Data Sources UI" tab with per-database Browser (catalog + sample data), Query Console (SqlWorkbench) and embedded interactive Shell (DbShell via SSH). Backend: - Extend sql_console.py with Cassandra (CQL) + Neo4j (Cypher) engines - Add GET /api/sql/catalog/{engine} and GET /api/sql/sample/{engine} - ssh_terminal: optional initial_command for auto-launching DB CLIs Frontend: - DataSourcesView with 5-DB rail, health dots, Browser/Console/Shell sub-tabs - DbShell embedded xterm terminal with docker exec CLI per engine - Deep-link topology DB nodes to Data Sources UI (no SQL dock on platform) - WorkbenchPanel restricted to agent mode only — frees dashboard space --- api/sql_console.py | 356 ++++++++++++++++-- api/ssh_terminal.py | 7 + ui/src/App.tsx | 10 +- .../components/features/DataSourcesView.tsx | 311 +++++++++++++++ ui/src/components/features/DbShell.tsx | 210 +++++++++++ ui/src/components/features/SqlWorkbench.tsx | 23 +- ui/src/components/features/WorkbenchPanel.tsx | 13 +- ui/src/components/layout/SideNav.tsx | 5 +- ui/src/hooks/useCommandCenter.ts | 32 +- ui/src/lib/dataSourceCatalog.ts | 136 +++++++ 10 files changed, 1027 insertions(+), 76 deletions(-) create mode 100644 ui/src/components/features/DataSourcesView.tsx create mode 100644 ui/src/components/features/DbShell.tsx create mode 100644 ui/src/lib/dataSourceCatalog.ts diff --git a/api/sql_console.py b/api/sql_console.py index 471c7f3..ad7a5b0 100644 --- a/api/sql_console.py +++ b/api/sql_console.py @@ -1,4 +1,4 @@ -"""Live SQL console — PostgreSQL, MySQL, MongoDB + Trino with benchmark.""" +"""Live SQL console — PostgreSQL, MySQL, MongoDB, Cassandra, Neo4j + Trino.""" from __future__ import annotations @@ -10,8 +10,11 @@ from typing import Any import httpx import psycopg2 import pymysql -from fastapi import APIRouter +from cassandra.auth import PlainTextAuthProvider +from cassandra.cluster import Cluster +from fastapi import APIRouter, Query from fastapi.responses import JSONResponse +from neo4j import GraphDatabase from pydantic import BaseModel, Field from pymongo import MongoClient @@ -30,9 +33,21 @@ MONGO_HOST = os.getenv("MONGO_HOST", DB_HOST) MONGO_PORT = int(os.getenv("MONGO_PORT", "27017")) MONGO_DB = os.getenv("MONGO_DATABASE", "supplychain") +CASS_PORT = int(os.getenv("CASSANDRA_PORT", "9042")) +CASS_USER = os.getenv("CASSANDRA_USER", "cassandra") +CASS_PASS = os.getenv("CASSANDRA_PASSWORD", "cassandra") +CASS_KS = os.getenv("CASSANDRA_KEYSPACE", "telemetry") + +NEO4J_URI = os.getenv("NEO4J_URI", f"bolt://{DB_HOST}:7687") +NEO4J_USER = os.getenv("NEO4J_USER", "neo4j") +NEO4J_PASS = os.getenv("NEO4J_PASSWORD", "testpwd") + TRINO_URL = os.getenv("TRINO_URL", "http://10.0.21.50:8089").rstrip("/") TRINO_USER = os.getenv("TRINO_USER", "atc") +SOURCE_ENGINES = ("postgres", "mysql", "mongodb", "cassandra", "neo4j") +ENGINES = SOURCE_ENGINES + ("trino",) + router = APIRouter(prefix="/api/sql", tags=["sql"]) SAMPLES: dict[str, list[dict[str, str]]] = { @@ -45,7 +60,7 @@ SAMPLES: dict[str, list[dict[str, str]]] = { {"id": "pg6", "label": "Database sizes", "sql": "SELECT datname, pg_size_pretty(pg_database_size(datname)) AS size FROM pg_database ORDER BY pg_database_size(datname) DESC LIMIT 10;"}, {"id": "pg7", "label": "Row counts (stats)", "sql": "SELECT relname, n_live_tup FROM pg_stat_user_tables ORDER BY n_live_tup DESC NULLS LAST LIMIT 10;"}, {"id": "pg8", "label": "Connections", "sql": "SELECT count(*) AS connections, state FROM pg_stat_activity GROUP BY state;"}, - {"id": "pg9", "label": "Explain scan 100k", "sql": "EXPLAIN ANALYZE SELECT count(*) FROM generate_series(1, 100000);"}, + {"id": "pg9", "label": "Sales orders sample", "sql": "SELECT order_id, customer_id, region, total_amount FROM sales_orders ORDER BY order_id DESC LIMIT 10;"}, {"id": "pg10", "label": "Memory settings", "sql": "SELECT name, setting, unit FROM pg_settings WHERE name IN ('max_connections','shared_buffers','work_mem','effective_cache_size');"}, ], "mysql": [ @@ -72,6 +87,30 @@ SAMPLES: dict[str, list[dict[str, str]]] = { {"id": "mg9", "label": "Indexes on events", "sql": "INDEXES supplychain.events"}, {"id": "mg10", "label": "Recent events", "sql": "FIND supplychain.events SORT timestamp DESC LIMIT 5"}, ], + "cassandra": [ + {"id": "cs1", "label": "Cluster release", "sql": "SELECT release_version FROM system.local"}, + {"id": "cs2", "label": "Keyspaces", "sql": "SELECT keyspace_name FROM system_schema.keyspaces"}, + {"id": "cs3", "label": "Telemetry tables", "sql": "SELECT table_name FROM system_schema.tables WHERE keyspace_name='telemetry'"}, + {"id": "cs4", "label": "Device metrics sample", "sql": "SELECT device_id, metric_name, metric_value, recorded_at FROM telemetry.device_metrics LIMIT 10"}, + {"id": "cs5", "label": "Metrics by device", "sql": "SELECT device_id, count(*) AS cnt FROM telemetry.device_metrics GROUP BY device_id LIMIT 10"}, + {"id": "cs6", "label": "Distinct metric names", "sql": "SELECT DISTINCT metric_name FROM telemetry.device_metrics LIMIT 20"}, + {"id": "cs7", "label": "Table columns", "sql": "SELECT column_name, type FROM system_schema.columns WHERE keyspace_name='telemetry' AND table_name='device_metrics'"}, + {"id": "cs8", "label": "Recent metrics", "sql": "SELECT device_id, metric_name, metric_value FROM telemetry.device_metrics LIMIT 15"}, + {"id": "cs9", "label": "Partition token sample", "sql": "SELECT token(device_id) AS tok, device_id FROM telemetry.device_metrics LIMIT 5"}, + {"id": "cs10", "label": "Cluster peers", "sql": "SELECT peer, data_center, rack FROM system.peers"}, + ], + "neo4j": [ + {"id": "nj1", "label": "Node labels", "sql": "CALL db.labels() YIELD label RETURN label"}, + {"id": "nj2", "label": "Relationship types", "sql": "CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType"}, + {"id": "nj3", "label": "Total nodes", "sql": "MATCH (n) RETURN count(n) AS node_count"}, + {"id": "nj4", "label": "Total relationships", "sql": "MATCH ()-[r]->() RETURN count(r) AS rel_count"}, + {"id": "nj5", "label": "Products sample", "sql": "MATCH (p:Product) RETURN p.product_id AS id, p.name AS name, p.category AS category LIMIT 10"}, + {"id": "nj6", "label": "Suppliers sample", "sql": "MATCH (s:Supplier) RETURN s.supplier_id AS id, s.name AS name, s.country AS country LIMIT 10"}, + {"id": "nj7", "label": "Supply chain links", "sql": "MATCH (s:Supplier)-[r:SUPPLIES]->(p:Product) RETURN s.name AS supplier, p.name AS product LIMIT 10"}, + {"id": "nj8", "label": "Products by category", "sql": "MATCH (p:Product) RETURN p.category AS category, count(*) AS cnt ORDER BY cnt DESC LIMIT 10"}, + {"id": "nj9", "label": "Schema visualization", "sql": "CALL db.schema.visualization()"}, + {"id": "nj10", "label": "Constraint info", "sql": "SHOW CONSTRAINTS"}, + ], "trino": [ {"id": "tq1", "label": "Trino version", "sql": "SELECT version()"}, {"id": "tq2", "label": "Catalogs", "sql": "SHOW CATALOGS"}, @@ -104,6 +143,12 @@ def _tabular(columns: list[str], rows: list[list[Any]], elapsed_ms: int, **extra } +def _fmt(val: Any) -> Any: + if val is None or isinstance(val, (str, int, float, bool)): + return val + return str(val)[:200] + + def _run_postgres(sql: str, limit: int = 200) -> dict[str, Any]: t0 = time.perf_counter() conn = psycopg2.connect( @@ -116,7 +161,7 @@ def _run_postgres(sql: str, limit: int = 200) -> dict[str, Any]: elapsed_ms = int((time.perf_counter() - t0) * 1000) if cur.description: columns = [d[0] for d in cur.description] - rows = [list(r) for r in cur.fetchmany(limit)] + rows = [[_fmt(r) for r in row] for row in cur.fetchmany(limit)] return _tabular(columns, rows, elapsed_ms, truncated=len(rows) >= limit) return _tabular([], [], elapsed_ms, message="OK") finally: @@ -135,7 +180,7 @@ def _run_mysql(sql: str, limit: int = 200) -> dict[str, Any]: elapsed_ms = int((time.perf_counter() - t0) * 1000) if cur.description: columns = [d[0] for d in cur.description] - rows = [list(r) for r in cur.fetchmany(limit)] + rows = [[_fmt(r) for r in row] for row in cur.fetchmany(limit)] return _tabular(columns, rows, elapsed_ms, truncated=len(rows) >= limit) return _tabular([], [], elapsed_ms, message="OK") finally: @@ -146,14 +191,6 @@ def _mongo_client() -> MongoClient: return MongoClient(f"mongodb://{MONGO_HOST}:{MONGO_PORT}/", serverSelectionTimeoutMS=8000) -def _fmt_mongo_value(val: Any) -> Any: - if val is None: - return None - if isinstance(val, (str, int, float, bool)): - return val - return str(val)[:200] - - def _run_mongo(query: str, limit: int = 200) -> dict[str, Any]: t0 = time.perf_counter() q = query.strip() @@ -185,12 +222,12 @@ def _run_mongo(query: str, limit: int = 200) -> dict[str, Any]: rows = [["(empty)"]] else: columns = sorted({k for d in docs for k in d if k != "_id"}) - rows = [[_fmt_mongo_value(d.get(c)) for c in columns] for d in docs] + rows = [[_fmt(d.get(c)) for c in columns] for d in docs] elif m := re.match(r"^DISTINCT\s+(\w+)\.(\w+)\s+(\w+)$", q, re.I): db_name, coll, field = m.group(1), m.group(2), m.group(3) columns = [field] vals = client[db_name][coll].distinct(field)[:limit] - rows = [[_fmt_mongo_value(v)] for v in vals] + rows = [[_fmt(v)] for v in vals] elif m := re.match(r"^AGGREGATE\s+(\w+)\.(\w+)\s+GROUP\s+(\w+)\s+TOP\s+(\d+)$", q, re.I): db_name, coll, field, top_n = m.group(1), m.group(2), m.group(3), int(m.group(4)) pipe = [ @@ -222,6 +259,47 @@ def _run_mongo(query: str, limit: int = 200) -> dict[str, Any]: client.close() +def _cass_cluster() -> Cluster: + auth = PlainTextAuthProvider(CASS_USER, CASS_PASS) if CASS_USER else None + return Cluster([DB_HOST], port=CASS_PORT, auth_provider=auth, connect_timeout=8) + + +def _run_cassandra(cql: str, limit: int = 200) -> dict[str, Any]: + t0 = time.perf_counter() + cluster = _cass_cluster() + session = cluster.connect() + try: + rows_raw = list(session.execute(cql, timeout=30)) + elapsed_ms = int((time.perf_counter() - t0) * 1000) + if not rows_raw: + return _tabular([], [], elapsed_ms, message="OK") + columns = list(rows_raw[0]._fields) + rows = [[_fmt(getattr(r, c)) for c in columns] for r in rows_raw[:limit]] + return _tabular(columns, rows, elapsed_ms, truncated=len(rows_raw) > limit) + finally: + cluster.shutdown() + + +def _neo4j_driver(): + return GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASS)) + + +def _run_neo4j(cypher: str, limit: int = 200) -> dict[str, Any]: + t0 = time.perf_counter() + driver = _neo4j_driver() + try: + with driver.session() as session: + result = session.run(cypher) + keys = result.keys() + columns = list(keys) + rows_raw = [record.values() for record in result] + elapsed_ms = int((time.perf_counter() - t0) * 1000) + rows = [[_fmt(v) for v in row] for row in rows_raw[:limit]] + return _tabular(columns, rows, elapsed_ms, truncated=len(rows_raw) > limit) + finally: + driver.close() + + def _run_trino(sql: str, limit: int = 200) -> dict[str, Any]: t0 = time.perf_counter() headers = {"X-Trino-User": TRINO_USER, "Content-Type": "text/plain"} @@ -253,14 +331,210 @@ def _run_trino(sql: str, limit: int = 200) -> dict[str, Any]: return _tabular(columns, rows, elapsed_ms, engine_stats_ms=stats.get("elapsedTimeMillis"), truncated=len(rows) >= limit) -ENGINES = ("postgres", "mysql", "mongodb", "trino") - - class SqlRequest(BaseModel): - engine: str = Field(..., pattern="^(postgres|mysql|mongodb|trino)$") + engine: str = Field(..., pattern="^(postgres|mysql|mongodb|trino|cassandra|neo4j)$") sql: str = Field(..., min_length=1, max_length=8000) +def _connection_info(engine: str) -> dict[str, str]: + if engine == "postgres": + return {"host": DB_HOST, "port": str(PG_PORT), "database": PG_DB, "user": PG_USER} + if engine == "mysql": + return {"host": DB_HOST, "port": str(MYSQL_PORT), "database": MYSQL_DB, "user": MYSQL_USER} + if engine == "mongodb": + return {"host": MONGO_HOST, "port": str(MONGO_PORT), "database": MONGO_DB} + if engine == "cassandra": + return {"host": DB_HOST, "port": str(CASS_PORT), "keyspace": CASS_KS, "user": CASS_USER} + if engine == "neo4j": + return {"uri": NEO4J_URI, "user": NEO4J_USER} + return {"url": TRINO_URL, "user": TRINO_USER} + + +def _dispatch(engine: str, sql: str, limit: int = 200) -> dict[str, Any]: + if engine == "postgres": + return _run_postgres(sql, limit) + if engine == "mysql": + return _run_mysql(sql, limit) + if engine == "mongodb": + return _run_mongo(sql, limit) + if engine == "cassandra": + return _run_cassandra(sql, limit) + if engine == "neo4j": + return _run_neo4j(sql, limit) + return _run_trino(sql, limit) + + +def _catalog_postgres() -> dict[str, Any]: + conn = psycopg2.connect(host=DB_HOST, port=PG_PORT, user=PG_USER, password=PG_PASS, dbname=PG_DB, connect_timeout=8) + try: + conn.set_session(readonly=True, autocommit=True) + cur = conn.cursor() + cur.execute( + "SELECT schemaname, relname, n_live_tup FROM pg_stat_user_tables ORDER BY n_live_tup DESC NULLS LAST LIMIT 50" + ) + objects = [] + for schema, table, rows in cur.fetchall(): + objects.append({"type": "table", "schema": schema, "name": table, "fqn": f"{schema}.{table}", "row_count": int(rows or 0)}) + cur.execute("SELECT version()") + version = cur.fetchone()[0] + return {"engine": "postgres", "version": version, "objects": objects} + finally: + conn.close() + + +def _catalog_mysql() -> dict[str, Any]: + conn = pymysql.connect(host=DB_HOST, port=MYSQL_PORT, user=MYSQL_USER, password=MYSQL_PASS, database=MYSQL_DB, connect_timeout=8) + try: + cur = conn.cursor() + cur.execute( + "SELECT table_name, table_rows, ROUND((data_length+index_length)/1024/1024,1) AS mb " + "FROM information_schema.tables WHERE table_schema=%s ORDER BY table_rows DESC", + (MYSQL_DB,), + ) + objects = [{"type": "table", "schema": MYSQL_DB, "name": r[0], "fqn": f"{MYSQL_DB}.{r[0]}", "row_count": int(r[1] or 0), "size_mb": float(r[2] or 0)} for r in cur.fetchall()] + cur.execute("SELECT VERSION()") + version = cur.fetchone()[0] + return {"engine": "mysql", "version": version, "objects": objects} + finally: + conn.close() + + +def _catalog_mongodb() -> dict[str, Any]: + client = _mongo_client() + try: + objects = [] + for db_name in client.list_database_names(): + if db_name in ("admin", "config", "local"): + continue + for coll in client[db_name].list_collection_names(): + try: + cnt = client[db_name][coll].estimated_document_count() + except Exception: + cnt = None + objects.append({"type": "collection", "schema": db_name, "name": coll, "fqn": f"{db_name}.{coll}", "row_count": cnt}) + build = client.admin.command("buildInfo") + return {"engine": "mongodb", "version": build.get("version"), "objects": objects} + finally: + client.close() + + +def _catalog_cassandra() -> dict[str, Any]: + cluster = _cass_cluster() + session = cluster.connect() + try: + objects = [] + kss = [ + r.keyspace_name for r in session.execute("SELECT keyspace_name FROM system_schema.keyspaces") + if not r.keyspace_name.startswith("system") + ] + for ks in kss: + tables = session.execute( + "SELECT table_name FROM system_schema.tables WHERE keyspace_name=%s", (ks,) + ) + for t in tables: + objects.append({"type": "table", "schema": ks, "name": t.table_name, "fqn": f"{ks}.{t.table_name}", "row_count": None}) + ver = list(session.execute("SELECT release_version FROM system.local"))[0].release_version + return {"engine": "cassandra", "version": ver, "objects": objects} + finally: + cluster.shutdown() + + +def _catalog_neo4j() -> dict[str, Any]: + driver = _neo4j_driver() + try: + with driver.session() as session: + labels = [r["label"] for r in session.run("CALL db.labels() YIELD label RETURN label")] + rels = [r["relationshipType"] for r in session.run("CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType")] + objects = [] + for label in labels: + cnt = session.run(f"MATCH (n:`{label}`) RETURN count(n) AS c").single()["c"] + objects.append({"type": "node_label", "schema": "graph", "name": label, "fqn": label, "row_count": cnt}) + for rel in rels: + cnt = session.run(f"MATCH ()-[r:`{rel}`]->() RETURN count(r) AS c").single()["c"] + objects.append({"type": "relationship", "schema": "graph", "name": rel, "fqn": rel, "row_count": cnt}) + total = session.run("MATCH (n) RETURN count(n) AS c").single()["c"] + return {"engine": "neo4j", "version": "4.4", "total_nodes": total, "objects": objects} + finally: + driver.close() + + +def _catalog(engine: str) -> dict[str, Any]: + if engine == "postgres": + return _catalog_postgres() + if engine == "mysql": + return _catalog_mysql() + if engine == "mongodb": + return _catalog_mongodb() + if engine == "cassandra": + return _catalog_cassandra() + if engine == "neo4j": + return _catalog_neo4j() + raise ValueError(f"Catalog not supported for {engine}") + + +def _sample_postgres(object_name: str, limit: int) -> dict[str, Any]: + if "." in object_name: + schema, table = object_name.split(".", 1) + sql = f'SELECT * FROM "{schema}"."{table}" LIMIT {limit}' + else: + sql = f'SELECT * FROM public."{object_name}" LIMIT {limit}' + return _run_postgres(sql, limit) + + +def _sample_mysql(object_name: str, limit: int) -> dict[str, Any]: + table = object_name.split(".")[-1] + return _run_mysql(f"SELECT * FROM `{table}` LIMIT {limit}", limit) + + +def _sample_mongodb(object_name: str, limit: int) -> dict[str, Any]: + if "." in object_name: + db_name, coll = object_name.split(".", 1) + else: + db_name, coll = MONGO_DB, object_name + return _run_mongo(f"FIND {db_name}.{coll} LIMIT {limit}", limit) + + +def _sample_cassandra(object_name: str, limit: int) -> dict[str, Any]: + if "." in object_name: + ks, table = object_name.split(".", 1) + else: + ks, table = CASS_KS, object_name + return _run_cassandra(f"SELECT * FROM {ks}.{table} LIMIT {limit}", limit) + + +def _sample_neo4j(object_name: str, limit: int) -> dict[str, Any]: + label = object_name.split(".")[-1] + cypher = f"MATCH (n:`{label}`) RETURN n LIMIT {limit}" + t0 = time.perf_counter() + driver = _neo4j_driver() + try: + with driver.session() as session: + result = session.run(cypher) + docs = [dict(record["n"]) for record in result] + elapsed_ms = int((time.perf_counter() - t0) * 1000) + if not docs: + return _tabular(["result"], [["(empty)"]], elapsed_ms) + columns = sorted({k for d in docs for k in d}) + rows = [[_fmt(d.get(c)) for c in columns] for d in docs] + return _tabular(columns, rows, elapsed_ms, object=object_name) + finally: + driver.close() + + +def _sample(engine: str, object_name: str, limit: int) -> dict[str, Any]: + if engine == "postgres": + return _sample_postgres(object_name, limit) + if engine == "mysql": + return _sample_mysql(object_name, limit) + if engine == "mongodb": + return _sample_mongodb(object_name, limit) + if engine == "cassandra": + return _sample_cassandra(object_name, limit) + if engine == "neo4j": + return _sample_neo4j(object_name, limit) + raise ValueError(f"Sample not supported for {engine}") + + @router.get("/samples/{engine}") async def get_samples(engine: str): if engine not in SAMPLES: @@ -271,7 +545,7 @@ async def get_samples(engine: str): @router.get("/health") async def sql_health(): out: dict[str, Any] = {} - for eng in ENGINES: + for eng in SOURCE_ENGINES: try: if eng == "postgres": _run_postgres("SELECT 1") @@ -282,32 +556,38 @@ async def sql_health(): elif eng == "mongodb": _run_mongo("SHOW DATABASES") out[eng] = {"ok": True, "host": MONGO_HOST, "database": MONGO_DB, "error": None} + elif eng == "cassandra": + _run_cassandra("SELECT release_version FROM system.local LIMIT 1") + out[eng] = {"ok": True, "host": DB_HOST, "keyspace": CASS_KS, "error": None} else: - r = _run_trino("SELECT 1") - out[eng] = {"ok": bool(r.get("ok")), "url": TRINO_URL, "user": TRINO_USER, "error": r.get("error")} + _run_neo4j("RETURN 1 AS n") + out[eng] = {"ok": True, "uri": NEO4J_URI, "user": NEO4J_USER, "error": None} except Exception as exc: out[eng] = {"ok": False, "error": str(exc)[:200]} return out -def _connection_info(engine: str) -> dict[str, str]: - if engine == "postgres": - return {"host": DB_HOST, "port": str(PG_PORT), "database": PG_DB, "user": PG_USER} - if engine == "mysql": - return {"host": DB_HOST, "port": str(MYSQL_PORT), "database": MYSQL_DB, "user": MYSQL_USER} - if engine == "mongodb": - return {"host": MONGO_HOST, "port": str(MONGO_PORT), "database": MONGO_DB} - return {"url": TRINO_URL, "user": TRINO_USER} +@router.get("/catalog/{engine}") +async def get_catalog(engine: str): + if engine not in SOURCE_ENGINES: + return JSONResponse({"error": "unknown engine"}, status_code=404) + try: + return _catalog(engine) + except Exception as exc: + return JSONResponse({"error": str(exc)[:500]}, status_code=502) -def _dispatch(engine: str, sql: str) -> dict[str, Any]: - if engine == "postgres": - return _run_postgres(sql) - if engine == "mysql": - return _run_mysql(sql) - if engine == "mongodb": - return _run_mongo(sql) - return _run_trino(sql) +@router.get("/sample/{engine}") +async def get_sample(engine: str, object: str = Query(..., min_length=1), limit: int = Query(50, ge=1, le=200)): + if engine not in SOURCE_ENGINES: + return JSONResponse({"error": "unknown engine"}, status_code=404) + try: + result = _sample(engine, object, limit) + if not result.get("ok"): + return JSONResponse(result, status_code=422) + return {**result, "engine": engine, "object": object} + except Exception as exc: + return JSONResponse({"ok": False, "error": str(exc)[:500]}, status_code=502) @router.post("/execute") diff --git a/api/ssh_terminal.py b/api/ssh_terminal.py index 3df63d5..4d79ece 100644 --- a/api/ssh_terminal.py +++ b/api/ssh_terminal.py @@ -86,6 +86,13 @@ async def ssh_session(ws: WebSocket) -> None: await _send(ws, "connected", message="connected") + initial_cmd = cfg.get("initial_command") + if initial_cmd and chan and not chan.closed: + try: + chan.send(initial_cmd if initial_cmd.endswith("\n") else initial_cmd + "\n") + except Exception: + pass + async def pump_out() -> None: assert chan is not None while True: diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 4bb790a..3bf6a65 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -21,6 +21,7 @@ import { DataGenView } from './components/features/DataGenView' import { ChangesView } from './components/features/ChangesView' import { DataFlowView } from './components/features/DataFlowView' import { SearchView } from './components/features/SearchView' +import { DataSourcesView } from './components/features/DataSourcesView' import { SshTerminal } from './components/features/SshTerminal' import { TerminalDock } from './components/features/TerminalDock' import { WorkbenchPanel } from './components/features/WorkbenchPanel' @@ -38,6 +39,7 @@ export default function App() { const mainScrollRef = useRef(null) const isPlatform = cc.mainView === 'platform' + const isDataSources = cc.mainView === 'datasources' const openApprovals = () => { cc.setMainView('approvals') @@ -79,9 +81,9 @@ export default function App() { onOpenSsh={() => setSshOpen(true)} /> -
+
-
+
{isPlatform && ( <>
@@ -119,7 +121,7 @@ export default function App() { )} -
+
{cc.mainView === 'platform' ? ( + ) : cc.mainView === 'datasources' ? ( + ) : cc.mainView === 'datagen' ? ( cc.setMainView('platform')} /> ) : cc.mainView === 'changes' ? ( diff --git a/ui/src/components/features/DataSourcesView.tsx b/ui/src/components/features/DataSourcesView.tsx new file mode 100644 index 0000000..6ea5ba1 --- /dev/null +++ b/ui/src/components/features/DataSourcesView.tsx @@ -0,0 +1,311 @@ +import { useCallback, useEffect, useState } from 'react' +import { + Activity, + ChevronRight, + Database, + FolderTree, + Loader2, + RefreshCw, + Server, + Table2, + TerminalSquare, +} from 'lucide-react' +import { Badge } from '../ui/Badge' +import { DbShell } from './DbShell' +import { SqlWorkbench } from './SqlWorkbench' +import { + getSourceMeta, + SOURCE_CATALOG, + type CatalogObject, + type SourceEngine, + type SourceSubTab, +} from '../../lib/dataSourceCatalog' +import { cn } from '../../lib/utils' +import { subTabActive, subTabIdle } from '../../lib/tabActive' + +type HealthMap = Record +type CatalogResponse = { + engine: string + version?: string + total_nodes?: number + objects: CatalogObject[] +} +type SampleResponse = { + ok: boolean + columns?: string[] + rows?: unknown[][] + row_count?: number + elapsed_ms?: number + error?: string +} + +type Props = { + focusEngine?: SourceEngine | null +} + +const SUB_TABS: { id: SourceSubTab; label: string; icon: typeof FolderTree }[] = [ + { id: 'browser', label: 'Browser', icon: FolderTree }, + { id: 'console', label: 'Query Console', icon: Database }, + { id: 'shell', label: 'Shell', icon: TerminalSquare }, +] + +function fmtCount(n?: number | null) { + if (n == null) return '—' + if (n >= 1_000_000) return `${(n / 1_000_000).toFixed(1)}M` + if (n >= 1_000) return `${(n / 1_000).toFixed(1)}K` + return String(n) +} + +export function DataSourcesView({ focusEngine }: Props) { + const [active, setActive] = useState(focusEngine || 'postgres') + const [subTab, setSubTab] = useState('browser') + const [health, setHealth] = useState({}) + const [catalog, setCatalog] = useState(null) + const [catalogLoading, setCatalogLoading] = useState(false) + const [selectedObject, setSelectedObject] = useState(null) + const [sample, setSample] = useState(null) + const [sampleLoading, setSampleLoading] = useState(false) + + const meta = getSourceMeta(active) + + useEffect(() => { + if (focusEngine) setActive(focusEngine) + }, [focusEngine]) + + const loadHealth = useCallback(async () => { + try { + const r = await fetch('/api/sql/health') + if (r.ok) setHealth(await r.json()) + } catch { /* */ } + }, []) + + const loadCatalog = useCallback(async (engine: SourceEngine) => { + setCatalogLoading(true) + setCatalog(null) + setSelectedObject(null) + setSample(null) + try { + const r = await fetch(`/api/sql/catalog/${engine}`) + if (r.ok) { + const j: CatalogResponse = await r.json() + setCatalog(j) + const first = j.objects?.find((o) => o.type === 'table' || o.type === 'collection' || o.type === 'node_label') + if (first) setSelectedObject(first) + } + } catch { /* */ } finally { + setCatalogLoading(false) + } + }, []) + + const loadSample = useCallback(async (engine: SourceEngine, obj: CatalogObject) => { + setSampleLoading(true) + setSample(null) + try { + const r = await fetch(`/api/sql/sample/${engine}?object=${encodeURIComponent(obj.fqn)}&limit=50`) + const j = await r.json() + setSample(j) + } catch { + setSample({ ok: false, error: 'Sample API unavailable' }) + } finally { + setSampleLoading(false) + } + }, []) + + useEffect(() => { loadHealth() }, [loadHealth]) + useEffect(() => { + if (subTab === 'browser') loadCatalog(active) + }, [active, subTab, loadCatalog]) + + useEffect(() => { + if (selectedObject && subTab === 'browser') loadSample(active, selectedObject) + }, [selectedObject, active, subTab, loadSample]) + + const refreshAll = () => { + loadHealth() + if (subTab === 'browser') loadCatalog(active) + else if (selectedObject) loadSample(active, selectedObject) + } + + return ( +
+ {/* Header */} +
+
+

+ + Data Sources UI +

+

+ Enterprise data browser — schema exploration, query console & interactive shells for all source databases +

+
+ +
+ +
+ {/* Left rail — database cards */} + + + {/* Main panel */} +
+ {/* Engine header */} +
+
+

+ + {meta.label} + {catalog?.version && ( + v{catalog.version.split(' ')[0]?.slice(0, 20)} + )} +

+

+ {meta.host}:{meta.port} · {meta.database} · container {meta.container} +

+
+
+ {SUB_TABS.map(({ id, label, icon: Icon }) => ( + + ))} +
+
+ + {/* Sub-tab content */} +
+ {subTab === 'browser' && ( +
+ {/* Object tree */} +
+
+ Objects + {catalogLoading && } +
+
+ {catalog?.objects?.map((obj) => ( + + ))} + {!catalogLoading && !catalog?.objects?.length && ( +

No objects found

+ )} +
+
+ + {/* Sample data grid */} +
+
+ + {selectedObject ? ( + <>Sample: {selectedObject.fqn} + ) : 'Select an object'} + + {sampleLoading && } +
+
+ {sample?.ok && sample.columns && ( + <> +

+ {sample.row_count} rows · {sample.elapsed_ms}ms +

+ + + + {sample.columns.map((c) => )} + + + + {sample.rows?.map((row, i) => ( + + {row.map((cell, j) => ( + + ))} + + ))} + +
{c}
+ {cell === null || cell === undefined ? 'NULL' : String(cell)} +
+ + )} + {sample && !sample.ok && ( +

{sample.error || 'Failed to load sample'}

+ )} + {!selectedObject && !sampleLoading && ( +

Select a table, collection or label to preview data

+ )} +
+
+
+ )} + + {subTab === 'console' && ( + + )} + + {subTab === 'shell' && ( + + )} +
+
+
+
+ ) +} diff --git a/ui/src/components/features/DbShell.tsx b/ui/src/components/features/DbShell.tsx new file mode 100644 index 0000000..53ce9c7 --- /dev/null +++ b/ui/src/components/features/DbShell.tsx @@ -0,0 +1,210 @@ +import { useCallback, useEffect, useRef, useState } from 'react' +import { Loader2, Plug, RotateCcw, TerminalSquare } from 'lucide-react' +import { Terminal } from '@xterm/xterm' +import { FitAddon } from '@xterm/addon-fit' +import '@xterm/xterm/css/xterm.css' +import { cn } from '../../lib/utils' +import { subTabActive, subTabIdle } from '../../lib/tabActive' + +type Props = { + host?: string + port?: string + username?: string + initialCommand?: string + className?: string +} + +type ConnState = 'form' | 'connecting' | 'connected' | 'closed' + +export function DbShell({ + host = '10.0.21.51', + port = '22', + username = 'root', + initialCommand, + className, +}: Props) { + const [password, setPassword] = useState('') + const [state, setState] = useState('form') + const [statusMsg, setStatusMsg] = useState(null) + const [remember, setRemember] = useState(false) + + const termRef = useRef(null) + const term = useRef(null) + const fit = useRef(null) + const ws = useRef(null) + + const teardown = useCallback(() => { + try { ws.current?.close() } catch { /* */ } + ws.current = null + try { term.current?.dispose() } catch { /* */ } + term.current = null + fit.current = null + }, []) + + const connect = useCallback(() => { + if (!password) return + setState('connecting') + setStatusMsg(`Connecting to ${username}@${host}:${port}…`) + + const t = new Terminal({ + cursorBlink: true, + fontSize: 12, + fontFamily: 'ui-monospace, SFMono-Regular, Menlo, monospace', + theme: { background: '#0a0e14', foreground: '#d6deeb', cursor: '#7ee787' }, + }) + const fitAddon = new FitAddon() + t.loadAddon(fitAddon) + term.current = t + fit.current = fitAddon + + requestAnimationFrame(() => { + if (!termRef.current) return + t.open(termRef.current) + try { fitAddon.fit() } catch { /* */ } + + const proto = window.location.protocol === 'https:' ? 'wss' : 'ws' + const socket = new WebSocket(`${proto}://${window.location.host}/api/ws/ssh`) + ws.current = socket + + socket.onopen = () => { + socket.send(JSON.stringify({ + type: 'connect', + host, + port: Number(port) || 22, + username, + password, + cols: t.cols, + rows: t.rows, + initial_command: initialCommand, + })) + } + socket.onmessage = (ev) => { + let msg: { type?: string; data?: string; message?: string } + try { msg = JSON.parse(ev.data) } catch { return } + if (msg.type === 'data') { + t.write(msg.data || '') + } else if (msg.type === 'status') { + setStatusMsg(msg.message || null) + } else if (msg.type === 'connected') { + setState('connected') + setStatusMsg(null) + if (remember) sessionStorage.setItem('ds-ssh-pass', password) + t.focus() + } else if (msg.type === 'error') { + setState('closed') + setStatusMsg(msg.message || 'Error') + t.writeln(`\r\n\x1b[31m${msg.message}\x1b[0m`) + } else if (msg.type === 'closed') { + setState('closed') + t.writeln('\r\n\x1b[33m*** Session closed ***\x1b[0m') + } + } + socket.onclose = () => { + setState((s) => (s === 'connected' ? 'closed' : s)) + } + + t.onData((d) => { + if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ type: 'data', data: d })) + }) + }) + }, [host, port, username, password, initialCommand, remember]) + + const disconnect = useCallback(() => { + try { ws.current?.send(JSON.stringify({ type: 'disconnect' })) } catch { /* */ } + teardown() + setState('form') + setStatusMsg(null) + }, [teardown]) + + useEffect(() => { + const saved = sessionStorage.getItem('ds-ssh-pass') + if (saved) setPassword(saved) + }, []) + + useEffect(() => { + if (state !== 'connected' && state !== 'closed') return + const onResize = () => { + try { + fit.current?.fit() + const tt = term.current + if (tt && ws.current?.readyState === WebSocket.OPEN) { + ws.current.send(JSON.stringify({ type: 'resize', cols: tt.cols, rows: tt.rows })) + } + } catch { /* */ } + } + const ro = new ResizeObserver(onResize) + if (termRef.current) ro.observe(termRef.current) + window.addEventListener('resize', onResize) + onResize() + return () => { ro.disconnect(); window.removeEventListener('resize', onResize) } + }, [state]) + + useEffect(() => () => teardown(), [teardown]) + + return ( +
+
+ + + Interactive Shell + {state === 'connected' && ( + live + )} + {initialCommand && ( + + {initialCommand} + + )} + + {(state === 'connected' || state === 'closed') && ( + + )} +
+ + {state === 'form' ? ( +
{ e.preventDefault(); connect() }} + > +

+ SSH to {username}@{host}:{port} and launch the database CLI. + Password is used for this session only. +

+ + + +
+ ) : ( +
+ {state === 'connecting' && ( +
+ {statusMsg || 'Connecting…'} +
+ )} +
+
+ )} +
+ ) +} diff --git a/ui/src/components/features/SqlWorkbench.tsx b/ui/src/components/features/SqlWorkbench.tsx index 4960ad1..bbed4f5 100644 --- a/ui/src/components/features/SqlWorkbench.tsx +++ b/ui/src/components/features/SqlWorkbench.tsx @@ -1,5 +1,6 @@ import { useCallback, useEffect, useState } from 'react' import { Database, Loader2, Play, Zap } from 'lucide-react' +import type { SourceEngine } from '../../lib/dataSourceCatalog' import { cn } from '../../lib/utils' import { subTabActive, subTabIdle } from '../../lib/tabActive' @@ -14,7 +15,7 @@ type SqlResult = { sql?: string } -type Engine = 'postgres' | 'mysql' | 'mongodb' | 'trino' +type Engine = SourceEngine | 'trino' type Props = { engine: Engine @@ -25,7 +26,7 @@ const ENGINE_META: Record([]) - const [sql, setSql] = useState('SELECT version();') + const [sql, setSql] = useState('') const [result, setResult] = useState(null) const [benchmark, setBenchmark] = useState<{ comparison?: { postgres_ms?: number; trino_ms?: number; faster?: string; speedup_factor?: number } @@ -146,7 +159,7 @@ export function SqlWorkbench({ engine, compact }: Props) {