"""Dataset Builder V5 resource for the CorePlexML SDK."""
from __future__ import annotations
from coreplexml._http import HTTPClient
[docs]
class BuilderResource:
"""Prepare AI-ready datasets through deterministic, auditable steps."""
def __init__(self, http: HTTPClient):
self._http = http
[docs]
def list_sessions(self, limit: int = 50) -> dict:
return self._http.get("/api/builder/sessions", params={"limit": limit})
[docs]
def create_session(
self,
*,
file_path: str | None = None,
dataset_id: str | None = None,
project_id: str | None = None,
use_case: str = "autodetect",
use_case_description: str | None = None,
auto_advance: bool = True,
) -> dict:
"""Create a session from either a local file or an existing dataset."""
if not file_path and not dataset_id:
raise ValueError("file_path or dataset_id is required")
fields: dict = {
"use_case": use_case,
"auto_advance": str(bool(auto_advance)).lower(),
}
if dataset_id:
fields["dataset_id"] = dataset_id
if project_id:
fields["project_id"] = project_id
if use_case_description:
fields["use_case_description"] = use_case_description
if file_path:
return self._http.upload("/api/builder/sessions", file_path, fields=fields)
return self._http.post_form("/api/builder/sessions", data=fields)
[docs]
def get_session(self, session_id: str) -> dict:
return self._http.get(f"/api/builder/sessions/{session_id}")
[docs]
def chat(self, session_id: str, message: str) -> dict:
return self._http.post(
f"/api/builder/sessions/{session_id}/chat",
json={"message": message},
)
[docs]
def propose_step(self, session_id: str, step: str) -> dict:
return self._http.post(
f"/api/builder/sessions/{session_id}/steps/{step}/propose"
)
[docs]
def execute_step(self, session_id: str, step: str) -> dict:
return self._http.post(
f"/api/builder/sessions/{session_id}/steps/{step}/execute"
)
[docs]
def upload_result(self, session_id: str, step_run_id: str, file_path: str) -> dict:
return self._http.upload(
f"/api/builder/sessions/{session_id}/upload-result",
file_path,
fields={"step_run_id": step_run_id},
)
[docs]
def finalize(
self,
session_id: str,
name: str,
*,
description: str = "",
format: str = "parquet",
project_id: str | None = None,
) -> dict:
body = {"name": name, "description": description, "format": format}
if project_id:
body["project_id"] = project_id
return self._http.post(
f"/api/builder/sessions/{session_id}/finalize", json=body
)