ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(data-recipe): split recipe backend routes for readability (seed/validate/jobs)

Browse source
This commit is contained in:
Shine1i 2026-02-26 14:05:32 +01:00
commit e4b64f3cd5
4 changed files with 279 additions and 246 deletions
Download patch file Download diff file Expand all files Collapse all files

23
studio/backend/routes/data_recipe/__init__.py Normal file
View file

@ -0,0 +1,23 @@
"""Data Recipe route package."""
from __future__ import annotations
import sys
from pathlib import Path
from fastapi import APIRouter
backend_path = Path(__file__).parent.parent.parent
if str(backend_path) not in sys.path:
sys.path.insert(0, str(backend_path))
from .jobs import router as jobs_router
from .seed import router as seed_router
from .validate import router as validate_router
router = APIRouter()
router.include_router(seed_router)
router.include_router(validate_router)
router.include_router(jobs_router)
__all__ = ["router"]

143
studio/backend/routes/data_recipe/jobs.py Normal file
View file

@ -0,0 +1,143 @@
"""Job lifecycle endpoints for data recipe."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import ValidationError
from core.data_recipe.jobs import get_job_manager
from models.data_recipe import JobCreateResponse, RecipePayload
router = APIRouter()
@router.post("/jobs", response_class=JSONResponse, response_model=JobCreateResponse)
def create_job(payload: RecipePayload):
recipe = payload.recipe
if not recipe.get("columns"):
raise HTTPException(status_code=400, detail="Recipe must include columns.")
run: dict[str, Any] = payload.run or {}
run.pop("artifact_path", None)
run.pop("dataset_name", None)
execution_type = str(run.get("execution_type") or "full").strip().lower()
if execution_type not in {"preview", "full"}:
raise HTTPException(status_code=400, detail="invalid execution_type: must be 'preview' or 'full'")
run["execution_type"] = execution_type
run_config_raw = run.get("run_config")
if run_config_raw is not None:
try:
from data_designer.config.run_config import RunConfig
RunConfig.model_validate(run_config_raw)
except (ImportError, ValidationError, TypeError, ValueError) as exc:
raise HTTPException(status_code=400, detail=f"invalid run_config: {exc}") from exc
mgr = get_job_manager()
try:
job_id = mgr.start(recipe=recipe, run=run)
except RuntimeError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"job_id": job_id}
@router.get("/jobs/{job_id}/status")
def job_status(job_id: str):
mgr = get_job_manager()
state = mgr.get_status(job_id)
if state is None:
raise HTTPException(status_code=404, detail="job not found")
return state
@router.get("/jobs/current")
def current_job():
mgr = get_job_manager()
state = mgr.get_current_status()
if state is None:
raise HTTPException(status_code=404, detail="no job")
return state
@router.post("/jobs/{job_id}/cancel")
def cancel_job(job_id: str):
mgr = get_job_manager()
ok = mgr.cancel(job_id)
if not ok:
raise HTTPException(status_code=404, detail="job not found")
return mgr.get_status(job_id)
@router.get("/jobs/{job_id}/analysis")
def job_analysis(job_id: str):
mgr = get_job_manager()
analysis = mgr.get_analysis(job_id)
if analysis is None:
raise HTTPException(status_code=404, detail="analysis not ready")
return analysis
@router.get("/jobs/{job_id}/dataset")
def job_dataset(
job_id: str,
limit: int = Query(default=20, ge=1, le=500),
offset: int = Query(default=0, ge=0),
):
mgr = get_job_manager()
result = mgr.get_dataset(job_id, limit=limit, offset=offset)
if result is None:
raise HTTPException(status_code=404, detail="dataset not ready")
if "error" in result:
raise HTTPException(status_code=422, detail=result["error"])
return {
"dataset": result["dataset"],
"total": result["total"],
"limit": limit,
"offset": offset,
}
@router.get("/jobs/{job_id}/events")
async def job_events(request: Request, job_id: str):
mgr = get_job_manager()
last_id = request.headers.get("last-event-id")
after_seq: int | None = None
if last_id:
try:
after_seq = int(str(last_id).strip())
except (TypeError, ValueError):
after_seq = None
after_q = request.query_params.get("after")
if after_q:
try:
after_seq = int(str(after_q).strip())
except (TypeError, ValueError):
pass
sub = mgr.subscribe(job_id, after_seq=after_seq)
if sub is None:
raise HTTPException(status_code=404, detail="job not found")
async def gen():
try:
for event in sub.replay:
yield sub.format_sse(event)
while True:
if await request.is_disconnected():
break
event = await sub.next_event(timeout_sec=1.0)
if event is None:
continue
yield sub.format_sse(event)
finally:
mgr.unsubscribe(sub)
return StreamingResponse(gen(), media_type="text/event-stream")

269
studio/backend/routes/data_recipe.py → studio/backend/routes/data_recipe/seed.py
View file

@ -1,42 +1,24 @@
"""
Data Recipe routes (DataDesigner runner).
"""
"""Seed inspect endpoints for data recipe."""
from __future__ import annotations
import base64
import binascii
import sys
from itertools import islice
from pathlib import Path
from typing import Any
from uuid import uuid4
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi import APIRouter, HTTPException
# same thing as other files do
backend_path = Path(__file__).parent.parent.parent
if str(backend_path) not in sys.path:
sys.path.insert(0, str(backend_path))
from core.data_recipe.jobs import get_job_manager
from core.data_recipe.service import (
build_config_builder,
create_data_designer,
validate_recipe,
)
from models.data_recipe import (
JobCreateResponse,
RecipePayload,
SeedInspectRequest,
SeedInspectUploadRequest,
SeedInspectResponse,
ValidateError,
ValidateResponse,
SeedInspectUploadRequest,
)
router = APIRouter()
DATA_EXTS = (".parquet", ".jsonl", ".json", ".csv")
DEFAULT_SPLIT = "train"
LOCAL_UPLOAD_EXTS = {".csv", ".json", ".jsonl"}
@ -70,20 +52,21 @@ def _normalize_optional_text(value: str | None) -> str | None:
def _list_hf_data_files(*, dataset_name: str, token: str | None) -> list[str]:
try:
from huggingface_hub import HfApi
from huggingface_hub.utils import HfHubHTTPError
except ImportError:
return []
try:
api = HfApi()
repo_files = api.list_repo_files(dataset_name, repo_type="dataset", token=token)
return [file for file in repo_files if file.lower().endswith(DATA_EXTS)]
except Exception:
except (HfHubHTTPError, OSError, ValueError):
return []
def _select_best_file(data_files: list[str], split: str | None) -> str | None:
def _select_best_file(data_files: list[str]) -> str | None:
if not data_files:
return None
if not split:
return data_files[0]
split_lower = split.lower()
split_lower = DEFAULT_SPLIT
def score(path: str) -> tuple[int, int]:
name = path.lower()
@ -102,8 +85,8 @@ def _select_best_file(data_files: list[str], split: str | None) -> str | None:
return sorted(data_files, key=score)[0]
def _resolve_seed_hf_path(dataset_name: str, data_files: list[str], split: str | None) -> str | None:
selected = _select_best_file(data_files, split)
def _resolve_seed_hf_path(dataset_name: str, data_files: list[str]) -> str | None:
selected = _select_best_file(data_files)
if not selected:
return None
@ -177,7 +160,7 @@ def _decode_base64_payload(content_base64: str) -> bytes:
def _read_preview_rows_from_local_file(path: Path, preview_size: int) -> list[dict[str, Any]]:
try:
import pandas as pd
except Exception as exc:
except ImportError as exc:
raise HTTPException(status_code=500, detail=f"seed inspect dependencies unavailable: {exc}") from exc
ext = path.suffix.lower()
@ -189,68 +172,19 @@ def _read_preview_rows_from_local_file(path: Path, preview_size: int) -> list[di
elif ext == ".json":
try:
df = pd.read_json(path, lines=True).head(preview_size)
except Exception:
except ValueError:
df = pd.read_json(path).head(preview_size)
else:
raise HTTPException(status_code=422, detail=f"unsupported file type: {ext}")
except HTTPException:
raise
except Exception as exc:
except (ValueError, OSError) as exc:
raise HTTPException(status_code=422, detail=f"seed inspect failed: {exc}") from exc
rows = df.to_dict(orient="records")
return _serialize_preview_rows(rows)
def _collect_validation_errors(recipe: dict[str, Any]) -> list[ValidateError]:
try:
from data_designer.engine.compiler import (
_add_internal_row_id_column_if_needed,
_get_allowed_references,
_resolve_and_add_seed_columns,
)
from data_designer.engine.validation import (
ViolationLevel,
validate_data_designer_config,
)
except Exception:
return []
try:
builder = build_config_builder(recipe)
designer = create_data_designer(recipe)
resource_provider = designer._create_resource_provider( # type: ignore[attr-defined]
"validate-configuration",
builder,
)
config = builder.build()
_resolve_and_add_seed_columns(config, resource_provider.seed_reader)
_add_internal_row_id_column_if_needed(config)
violations = validate_data_designer_config(
columns=config.columns,
processor_configs=config.processors or [],
allowed_references=_get_allowed_references(config),
)
except Exception:
return []
errors: list[ValidateError] = []
for violation in violations:
if violation.level != ViolationLevel.ERROR:
continue
code = getattr(violation.type, "value", None)
path = violation.column if violation.column else None
message = str(violation.message).strip() or "Validation failed."
errors.append(
ValidateError(
message=message,
path=path,
code=code,
)
)
return errors
@router.post("/seed/inspect", response_model=SeedInspectResponse)
def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
dataset_name = payload.dataset_name.strip()
@ -259,10 +193,10 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
try:
from datasets import load_dataset
except Exception as exc:
except ImportError as exc:
raise HTTPException(status_code=500, detail=f"seed inspect dependencies unavailable: {exc}") from exc
split = (payload.split or DEFAULT_SPLIT).strip() or DEFAULT_SPLIT
split = DEFAULT_SPLIT
subset = _normalize_optional_text(payload.subset)
token = _normalize_optional_text(payload.hf_token)
preview_size = int(payload.preview_size)
@ -270,7 +204,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
preview_rows: list[dict[str, Any]] = []
data_files = _list_hf_data_files(dataset_name=dataset_name, token=token)
selected_file = _select_best_file(data_files, split)
selected_file = _select_best_file(data_files)
if selected_file:
try:
single_file_kwargs = _build_stream_load_kwargs(
@ -285,7 +219,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
load_kwargs=single_file_kwargs,
preview_size=preview_size,
)
except Exception:
except (ValueError, OSError, RuntimeError):
preview_rows = []
if not preview_rows:
@ -301,7 +235,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
load_kwargs=split_kwargs,
preview_size=preview_size,
)
except Exception as exc:
except (ValueError, OSError, RuntimeError) as exc:
raise HTTPException(status_code=422, detail=f"seed inspect failed: {exc}") from exc
if not preview_rows:
@ -310,10 +244,9 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
columns = _extract_columns(preview_rows)
if not data_files:
# Best effort path fallback when file list is unavailable.
resolved_path = f"datasets/{dataset_name}/**/*.parquet"
else:
resolved_path = _resolve_seed_hf_path(dataset_name, data_files, split)
resolved_path = _resolve_seed_hf_path(dataset_name, data_files)
if not resolved_path:
raise HTTPException(status_code=422, detail="unable to resolve seed dataset path")
@ -322,7 +255,7 @@ def inspect_seed_dataset(payload: SeedInspectRequest) -> SeedInspectResponse:
resolved_path=resolved_path,
columns=columns,
preview_rows=preview_rows,
split=split,
split=None,
subset=subset,
)
@ -363,159 +296,3 @@ def inspect_seed_upload(payload: SeedInspectUploadRequest) -> SeedInspectRespons
split=None,
subset=None,
)
@router.post("/validate", response_model=ValidateResponse)
def validate(payload: RecipePayload) -> ValidateResponse:
recipe = payload.recipe
if not recipe.get("columns"):
return ValidateResponse(
valid=False,
errors=[ValidateError(message="Recipe must include columns.")],
)
try:
validate_recipe(recipe)
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except Exception as exc:
detail = str(exc).strip() or "Validation failed."
parsed_errors = _collect_validation_errors(recipe)
return ValidateResponse(
valid=False,
errors=parsed_errors or [ValidateError(message=detail)],
raw_detail=detail,
)
return ValidateResponse(valid=True)
@router.post("/jobs", response_class=JSONResponse, response_model=JobCreateResponse)
def create_job(payload: RecipePayload):
recipe = payload.recipe
if not recipe.get("columns"):
raise HTTPException(status_code=400, detail="Recipe must include columns.")
run: dict[str, Any] = payload.run or {}
run.pop("artifact_path", None)
run.pop("dataset_name", None)
execution_type = str(run.get("execution_type") or "full").strip().lower()
if execution_type not in {"preview", "full"}:
raise HTTPException(status_code=400, detail="invalid execution_type: must be 'preview' or 'full'")
run["execution_type"] = execution_type
run_config_raw = run.get("run_config")
if run_config_raw is not None:
try:
from data_designer.config.run_config import RunConfig
RunConfig.model_validate(run_config_raw)
except Exception as exc:
raise HTTPException(status_code=400, detail=f"invalid run_config: {exc}") from exc
mgr = get_job_manager()
try:
job_id = mgr.start(recipe=recipe, run=run)
except RuntimeError as exc:
raise HTTPException(status_code=409, detail=str(exc)) from exc
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
return {"job_id": job_id}
@router.get("/jobs/{job_id}/status")
def job_status(job_id: str):
mgr = get_job_manager()
state = mgr.get_status(job_id)
if state is None:
raise HTTPException(status_code=404, detail="job not found")
return state
@router.get("/jobs/current")
def current_job():
mgr = get_job_manager()
state = mgr.get_current_status()
if state is None:
raise HTTPException(status_code=404, detail="no job")
return state
@router.post("/jobs/{job_id}/cancel")
def cancel_job(job_id: str):
mgr = get_job_manager()
ok = mgr.cancel(job_id)
if not ok:
raise HTTPException(status_code=404, detail="job not found")
return mgr.get_status(job_id)
@router.get("/jobs/{job_id}/analysis")
def job_analysis(job_id: str):
mgr = get_job_manager()
analysis = mgr.get_analysis(job_id)
if analysis is None:
raise HTTPException(status_code=404, detail="analysis not ready")
return analysis
@router.get("/jobs/{job_id}/dataset")
def job_dataset(
job_id: str,
limit: int = Query(default=20, ge=1, le=500),
offset: int = Query(default=0, ge=0),
):
mgr = get_job_manager()
result = mgr.get_dataset(job_id, limit=limit, offset=offset)
if result is None:
raise HTTPException(status_code=404, detail="dataset not ready")
if "error" in result:
raise HTTPException(status_code=422, detail=result["error"])
return {
"dataset": result["dataset"],
"total": result["total"],
"limit": limit,
"offset": offset,
}
@router.get("/jobs/{job_id}/events")
async def job_events(request: Request, job_id: str):
mgr = get_job_manager()
last_id = request.headers.get("last-event-id")
after_seq: int | None = None
if last_id:
try:
after_seq = int(str(last_id).strip())
except Exception:
after_seq = None
# EventSource can't set custom headers on first connect after a full page refresh,
# so allow resume via query param too: /events?after=<seq>
after_q = request.query_params.get("after")
if after_q:
try:
after_seq = int(str(after_q).strip())
except Exception:
pass
sub = mgr.subscribe(job_id, after_seq=after_seq)
if sub is None:
raise HTTPException(status_code=404, detail="job not found")
async def gen():
try:
for event in sub.replay:
yield sub.format_sse(event)
while True:
if await request.is_disconnected():
break
event = await sub.next_event(timeout_sec=1.0)
if event is None:
continue
yield sub.format_sse(event)
finally:
mgr.unsubscribe(sub)
return StreamingResponse(gen(), media_type="text/event-stream")

90
studio/backend/routes/data_recipe/validate.py Normal file
View file

@ -0,0 +1,90 @@
"""Validation endpoints for data recipe."""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, HTTPException
from core.data_recipe.service import (
build_config_builder,
create_data_designer,
validate_recipe,
)
from models.data_recipe import RecipePayload, ValidateError, ValidateResponse
router = APIRouter()
def _collect_validation_errors(recipe: dict[str, Any]) -> list[ValidateError]:
try:
from data_designer.engine.compiler import (
_add_internal_row_id_column_if_needed,
_get_allowed_references,
_resolve_and_add_seed_columns,
)
from data_designer.engine.validation import (
ViolationLevel,
validate_data_designer_config,
)
except ImportError:
return []
try:
builder = build_config_builder(recipe)
designer = create_data_designer(recipe)
resource_provider = designer._create_resource_provider( # type: ignore[attr-defined]
"validate-configuration",
builder,
)
config = builder.build()
_resolve_and_add_seed_columns(config, resource_provider.seed_reader)
_add_internal_row_id_column_if_needed(config)
violations = validate_data_designer_config(
columns=config.columns,
processor_configs=config.processors or [],
allowed_references=_get_allowed_references(config),
)
except (TypeError, ValueError, AttributeError):
return []
errors: list[ValidateError] = []
for violation in violations:
if violation.level != ViolationLevel.ERROR:
continue
code = getattr(violation.type, "value", None)
path = violation.column if violation.column else None
message = str(violation.message).strip() or "Validation failed."
errors.append(
ValidateError(
message=message,
path=path,
code=code,
)
)
return errors
@router.post("/validate", response_model=ValidateResponse)
def validate(payload: RecipePayload) -> ValidateResponse:
recipe = payload.recipe
if not recipe.get("columns"):
return ValidateResponse(
valid=False,
errors=[ValidateError(message="Recipe must include columns.")],
)
try:
validate_recipe(recipe)
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc)) from exc
except Exception as exc:
detail = str(exc).strip() or "Validation failed."
parsed_errors = _collect_validation_errors(recipe)
return ValidateResponse(
valid=False,
errors=parsed_errors or [ValidateError(message=detail)],
raw_detail=detail,
)
return ValidateResponse(valid=True)