strepsiades
156 lines
5.2 KiB
Python
156 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Query
|
|
from fastapi.responses import FileResponse
|
|
from pydantic import BaseModel
|
|
from sync_state_machine.common.persistence import PersistenceBackend
|
|
|
|
from .config_models import PipelineUiConfig, list_ui_presets, preset_ui_config, file_ui_config
|
|
from sync_state_machine.config import get_preset_file_candidates
|
|
from .runner import PipelineRunnerService
|
|
|
|
|
|
class LogsResponse(BaseModel):
|
|
chunks: list[str]
|
|
next_index: int
|
|
|
|
|
|
class SqlQueryRequest(BaseModel):
|
|
sql: str
|
|
limit: int = 200
|
|
|
|
|
|
class SqlQueryResponse(BaseModel):
|
|
columns: list[str]
|
|
rows: list[dict[str, Any]]
|
|
error: str = ""
|
|
|
|
|
|
class LoadFileRequest(BaseModel):
|
|
path: str
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
app = FastAPI(title="Sync State Machine UI", version="0.1.0")
|
|
service = PipelineRunnerService()
|
|
web_root = Path(__file__).resolve().parent / "web"
|
|
|
|
@app.get("/")
|
|
async def index() -> FileResponse:
|
|
return FileResponse(str(web_root / "index.html"))
|
|
|
|
@app.get("/api/config/default", response_model=PipelineUiConfig)
|
|
async def get_default_config() -> PipelineUiConfig:
|
|
return service.get_default_config()
|
|
|
|
@app.get("/api/config/presets")
|
|
async def get_presets() -> dict:
|
|
data = []
|
|
for name in list_ui_presets():
|
|
data.append(
|
|
{
|
|
"name": name,
|
|
"candidates": [str(p) for p in get_preset_file_candidates(name)],
|
|
}
|
|
)
|
|
return {"presets": data}
|
|
|
|
@app.get("/api/config/preset/{preset_name}", response_model=PipelineUiConfig)
|
|
async def get_preset_config(preset_name: str) -> PipelineUiConfig:
|
|
try:
|
|
return preset_ui_config(preset_name)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
@app.post("/api/config/load-file", response_model=PipelineUiConfig)
|
|
async def load_config_file(payload: LoadFileRequest) -> PipelineUiConfig:
|
|
try:
|
|
return file_ui_config(payload.path)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@app.get("/api/config/current", response_model=PipelineUiConfig)
|
|
async def get_current_config() -> PipelineUiConfig:
|
|
return service.get_config()
|
|
|
|
@app.get("/api/config/visualized")
|
|
async def get_visualized_config() -> dict:
|
|
cfg = service.get_config()
|
|
return cfg.model_dump()
|
|
|
|
@app.post("/api/config/current", response_model=PipelineUiConfig)
|
|
async def set_current_config(config: PipelineUiConfig) -> PipelineUiConfig:
|
|
try:
|
|
return service.set_config(config)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@app.get("/api/status")
|
|
async def get_status() -> dict:
|
|
status = service.get_status()
|
|
return {
|
|
"running": status.running,
|
|
"started_at": status.started_at,
|
|
"ended_at": status.ended_at,
|
|
"last_error": status.last_error,
|
|
"last_stats": status.last_stats or {},
|
|
"current_log_file": status.current_log_file,
|
|
}
|
|
|
|
@app.post("/api/run")
|
|
async def run_pipeline(config: PipelineUiConfig) -> dict:
|
|
try:
|
|
status = await service.start(config)
|
|
return {
|
|
"running": status.running,
|
|
"started_at": status.started_at,
|
|
"current_log_file": status.current_log_file,
|
|
}
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
|
|
@app.post("/api/stop")
|
|
async def stop_pipeline() -> dict:
|
|
status = await service.stop()
|
|
return {
|
|
"running": status.running,
|
|
"ended_at": status.ended_at,
|
|
"last_error": status.last_error,
|
|
}
|
|
|
|
@app.get("/api/logs", response_model=LogsResponse)
|
|
async def get_logs(since: int = Query(default=0, ge=0)) -> LogsResponse:
|
|
chunks, next_idx = service.get_logs(since)
|
|
return LogsResponse(chunks=chunks, next_index=next_idx)
|
|
|
|
@app.get("/api/records/summary")
|
|
async def get_records_summary() -> dict:
|
|
cfg = service.get_config()
|
|
return PersistenceBackend.records_summary(
|
|
backend=cfg.persist.backend,
|
|
db_path=str(Path(cfg.persist.db_path)),
|
|
)
|
|
|
|
@app.post("/api/records/query", response_model=SqlQueryResponse)
|
|
async def run_records_query(payload: SqlQueryRequest) -> SqlQueryResponse:
|
|
sql = (payload.sql or "").strip()
|
|
if not sql:
|
|
return SqlQueryResponse(columns=[], rows=[], error="SQL is empty")
|
|
|
|
cfg = service.get_config()
|
|
try:
|
|
columns, rows = PersistenceBackend.query_records(
|
|
backend=cfg.persist.backend,
|
|
db_path=str(Path(cfg.persist.db_path)),
|
|
sql=sql,
|
|
limit=payload.limit,
|
|
)
|
|
return SqlQueryResponse(columns=columns, rows=rows, error="")
|
|
except Exception as exc:
|
|
return SqlQueryResponse(columns=[], rows=[], error=f"{type(exc).__name__}: {exc}")
|
|
|
|
return app
|