diff --git a/studio/backend/routes/data_recipe/__init__.py b/studio/backend/routes/data_recipe/__init__.py new file mode 100644 index 0000000000..dc0301d1c9 --- /dev/null +++ b/studio/backend/routes/data_recipe/__init__.py @@ -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"] diff --git a/studio/backend/routes/data_recipe/jobs.py b/studio/backend/routes/data_recipe/jobs.py new file mode 100644 index 0000000000..ffbded9474 --- /dev/null +++ b/studio/backend/routes/data_recipe/jobs.py @@ -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") diff --git a/studio/backend/routes/data_recipe.py b/studio/backend/routes/data_recipe/seed.py similarity index 53% rename from studio/backend/routes/data_recipe.py rename to studio/backend/routes/data_recipe/seed.py index 8a713f2cec..eb02ab2bbf 100644 --- a/studio/backend/routes/data_recipe.py +++ b/studio/backend/routes/data_recipe/seed.py @@ -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= - 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") diff --git a/studio/backend/routes/data_recipe/validate.py b/studio/backend/routes/data_recipe/validate.py new file mode 100644 index 0000000000..a8755f9410 --- /dev/null +++ b/studio/backend/routes/data_recipe/validate.py @@ -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)