291 lines
11 KiB
Python
291 lines
11 KiB
Python
import json
|
|
import os
|
|
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
|
|
from fastapi.responses import FileResponse
|
|
from sse_starlette.sse import EventSourceResponse
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from backend.db.database import get_db
|
|
from backend.db.models import JobModel
|
|
from backend.schemas.models import (
|
|
DryRunResponse,
|
|
JobResponse,
|
|
ProjectCardResponse,
|
|
ProjectCreate,
|
|
ProjectResponse,
|
|
ProjectUpdate,
|
|
TableCreate,
|
|
TableResponse,
|
|
TableUpdate,
|
|
)
|
|
from backend.services.generation_service import GenerationService
|
|
from backend.services.project_service import ProjectService
|
|
from backend.services.table_service import (
|
|
TableLockedError,
|
|
TableService,
|
|
TableValidationError,
|
|
model_to_table_dict,
|
|
)
|
|
from backend.engine.dag import get_referencing_tables
|
|
from backend.engine.generators import generate_dry_run_sample
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
|
|
def format_table_response(table, project_tables) -> TableResponse:
|
|
all_dicts = [model_to_table_dict(t) for t in project_tables]
|
|
ref_by = get_referencing_tables(table.name, all_dicts)
|
|
return TableResponse(
|
|
id=table.id,
|
|
project_id=table.project_id,
|
|
slug=getattr(table, "slug", "") or table.name,
|
|
name=table.name,
|
|
archetype=table.archetype, # type: ignore
|
|
status=table.status, # type: ignore
|
|
format=table.format, # type: ignore
|
|
row_limit=table.row_limit,
|
|
exclude_if=table.exclude_if,
|
|
fields=table.fields_data or [],
|
|
columns=table.columns_data or [],
|
|
data=table.rows_data or [],
|
|
referencing_tables=ref_by,
|
|
created_at=table.created_at,
|
|
updated_at=table.updated_at,
|
|
)
|
|
|
|
|
|
# --- Projects ---
|
|
|
|
@router.get("/projects", response_model=List[ProjectCardResponse])
|
|
async def list_projects(db: AsyncSession = Depends(get_db)):
|
|
return await ProjectService.list_projects(db)
|
|
|
|
|
|
@router.post("/projects", response_model=ProjectCardResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_project(data: ProjectCreate, db: AsyncSession = Depends(get_db)):
|
|
proj = await ProjectService.create_project(db, data)
|
|
cards = await ProjectService.list_projects(db)
|
|
for c in cards:
|
|
if c.id == proj.id:
|
|
return c
|
|
return ProjectCardResponse(
|
|
id=proj.id,
|
|
slug=proj.slug or proj.id,
|
|
name=proj.name,
|
|
description=proj.description,
|
|
seed=proj.seed,
|
|
table_count=0,
|
|
record_count=0,
|
|
aggregate_count=0,
|
|
reference_count=0,
|
|
valid_count=0,
|
|
created_at=proj.created_at,
|
|
updated_at=proj.updated_at,
|
|
)
|
|
|
|
|
|
@router.get("/projects/{project_id}")
|
|
async def get_project(project_id: str, db: AsyncSession = Depends(get_db)):
|
|
proj = await ProjectService.get_project(db, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
|
|
tables = await TableService.list_tables_by_project(db, proj.id)
|
|
table_responses = [format_table_response(t, tables) for t in tables]
|
|
|
|
return {
|
|
"id": proj.id,
|
|
"slug": proj.slug or proj.id,
|
|
"name": proj.name,
|
|
"description": proj.description,
|
|
"seed": proj.seed,
|
|
"tables": table_responses,
|
|
"created_at": proj.created_at,
|
|
"updated_at": proj.updated_at,
|
|
}
|
|
|
|
|
|
@router.put("/projects/{project_id}")
|
|
async def update_project(project_id: str, data: ProjectUpdate, db: AsyncSession = Depends(get_db)):
|
|
proj = await ProjectService.update_project(db, project_id, data)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return proj
|
|
|
|
|
|
@router.delete("/projects/{project_id}")
|
|
async def delete_project(project_id: str, db: AsyncSession = Depends(get_db)):
|
|
success = await ProjectService.delete_project(db, project_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return {"message": "Project deleted successfully"}
|
|
|
|
|
|
@router.get("/projects/{project_id}/export")
|
|
async def export_project(project_id: str, db: AsyncSession = Depends(get_db)):
|
|
data = await ProjectService.export_project(db, project_id)
|
|
if not data:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
return data
|
|
|
|
|
|
@router.post("/projects/import", status_code=status.HTTP_201_CREATED)
|
|
async def import_project(payload: dict, db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
project = await ProjectService.import_project(db, payload)
|
|
return {"id": project.id, "slug": project.slug, "name": project.name, "message": "Project imported successfully"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"Import failed: {str(e)}")
|
|
|
|
|
|
# --- Tables ---
|
|
|
|
@router.post("/projects/{project_id}/tables", response_model=TableResponse, status_code=status.HTTP_201_CREATED)
|
|
async def create_table(project_id: str, data: TableCreate, db: AsyncSession = Depends(get_db)):
|
|
proj = await ProjectService.get_project(db, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
try:
|
|
table = await TableService.create_table(db, proj.id, data)
|
|
project_tables = await TableService.list_tables_by_project(db, proj.id)
|
|
return format_table_response(table, project_tables)
|
|
except TableValidationError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.get("/tables/{table_id}", response_model=TableResponse)
|
|
async def get_table(table_id: str, db: AsyncSession = Depends(get_db)):
|
|
table = await TableService.get_table(db, table_id)
|
|
if not table:
|
|
raise HTTPException(status_code=404, detail="Table not found")
|
|
project_tables = await TableService.list_tables_by_project(db, table.project_id)
|
|
return format_table_response(table, project_tables)
|
|
|
|
|
|
@router.get("/projects/{project_id}/tables/{table_identifier}", response_model=TableResponse)
|
|
async def get_project_table(project_id: str, table_identifier: str, db: AsyncSession = Depends(get_db)):
|
|
proj = await ProjectService.get_project(db, project_id)
|
|
if not proj:
|
|
raise HTTPException(status_code=404, detail="Project not found")
|
|
table = await TableService.get_table(db, table_identifier, project_id=proj.id)
|
|
if not table:
|
|
raise HTTPException(status_code=404, detail="Table not found")
|
|
project_tables = await TableService.list_tables_by_project(db, proj.id)
|
|
return format_table_response(table, project_tables)
|
|
|
|
|
|
@router.put("/tables/{table_id}", response_model=TableResponse)
|
|
async def update_table(table_id: str, data: TableUpdate, db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
table = await TableService.update_table(db, table_id, data)
|
|
project_tables = await TableService.list_tables_by_project(db, table.project_id)
|
|
return format_table_response(table, project_tables)
|
|
except TableLockedError as e:
|
|
raise HTTPException(status_code=409, detail=str(e))
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=404, detail=str(e))
|
|
|
|
|
|
@router.delete("/tables/{table_id}")
|
|
async def delete_table(table_id: str, db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
success = await TableService.delete_table(db, table_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Table not found")
|
|
return {"message": "Table deleted successfully"}
|
|
except TableLockedError as e:
|
|
raise HTTPException(status_code=409, detail=str(e))
|
|
|
|
|
|
@router.post("/tables/{table_id}/validate", response_model=DryRunResponse)
|
|
async def validate_table(table_id: str, db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
result = await TableService.validate_table(db, table_id)
|
|
return DryRunResponse(success=True, sample_rows=result["sample_rows"])
|
|
except TableValidationError as e:
|
|
return DryRunResponse(success=False, error=str(e))
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=404, detail=str(e))
|
|
|
|
|
|
@router.post("/tables/{table_id}/invalidate")
|
|
async def invalidate_table(table_id: str, db: AsyncSession = Depends(get_db)):
|
|
try:
|
|
table = await TableService.invalidate_table(db, table_id)
|
|
return {"status": table.status}
|
|
except TableLockedError as e:
|
|
raise HTTPException(status_code=409, detail=str(e))
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=404, detail=str(e))
|
|
|
|
|
|
@router.get("/tables/{table_id}/preview", response_model=DryRunResponse)
|
|
async def preview_table(table_id: str, count: int = Query(10, le=50), db: AsyncSession = Depends(get_db)):
|
|
table = await TableService.get_table(db, table_id)
|
|
if not table:
|
|
raise HTTPException(status_code=404, detail="Table not found")
|
|
|
|
project_tables = await TableService.list_tables_by_project(db, table.project_id)
|
|
all_dicts = [model_to_table_dict(t) for t in project_tables]
|
|
ref_tables = [t for t in all_dicts if t.get("archetype") == "reference"]
|
|
|
|
try:
|
|
samples = generate_dry_run_sample(model_to_table_dict(table), ref_tables, sample_size=count)
|
|
return DryRunResponse(success=True, sample_rows=samples)
|
|
except Exception as e:
|
|
return DryRunResponse(success=False, error=f"{type(e).__name__}: {str(e)}")
|
|
|
|
|
|
# --- Generation & SSE Telemetry ---
|
|
|
|
@router.post("/projects/{project_id}/tables/{table_id}/generate", response_model=JobResponse)
|
|
async def start_generation(
|
|
project_id: str,
|
|
table_id: str,
|
|
format: Optional[str] = Query(None),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
try:
|
|
job = await GenerationService.create_job(db, project_id, table_id, format_override=format)
|
|
return job
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.get("/jobs/{job_id}", response_model=JobResponse)
|
|
async def get_job_status(job_id: str, db: AsyncSession = Depends(get_db)):
|
|
job = await GenerationService.get_job(db, job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="Job not found")
|
|
return job
|
|
|
|
|
|
@router.post("/jobs/{job_id}/cancel")
|
|
async def cancel_job(job_id: str, db: AsyncSession = Depends(get_db)):
|
|
cancelled = await GenerationService.cancel_job(db, job_id)
|
|
return {"cancelled": cancelled}
|
|
|
|
|
|
@router.get("/jobs/{job_id}/events")
|
|
async def stream_job_events(job_id: str):
|
|
async def event_generator():
|
|
async for event in GenerationService.subscribe(job_id):
|
|
yield {"event": "message", "data": json.dumps(event)}
|
|
|
|
return EventSourceResponse(event_generator())
|
|
|
|
|
|
@router.get("/artifacts/{job_id}")
|
|
async def download_artifact(job_id: str, db: AsyncSession = Depends(get_db)):
|
|
job = await GenerationService.get_job(db, job_id)
|
|
if not job or not job.artifact_path or not os.path.exists(job.artifact_path):
|
|
raise HTTPException(status_code=404, detail="Artifact file not found or generation not completed")
|
|
|
|
filename = f"{job.table_name}.{job.format}"
|
|
media_type = "text/csv" if job.format == "csv" else "application/octet-stream"
|
|
return FileResponse(
|
|
path=job.artifact_path,
|
|
filename=filename,
|
|
media_type=media_type,
|
|
)
|