Workflow | Mixtrain Docs

from mixtrain import Workflow

Constructor

Workflow(name: str)

Creates a reference to an existing workflow. This is a lazy operation - no API call is made until you access properties or call methods.

Parameter	Type	Description
`name`	`str`	Workflow name

workflow = Workflow("data-pipeline")

Property	Type	Description
`name`	`str`	Workflow name
`description`	`str`	Workflow description
`metadata`	`dict`	Full metadata dictionary (cached)
`runs`	`list`	List of runs

Run workflow synchronously. Blocks until completion.

workflow.run(config: dict = None) -> dict

Parameter	Type	Description
`config`	`dict`	Configuration parameters

Returns: dict with outputs and run metadata

result = workflow.run({"batch_size": 64, "epochs": 10})
print(result["outputs"])

Submit workflow asynchronously. Returns immediately.

workflow.submit(config: dict = None) -> dict

Parameter	Type	Description
`config`	`dict`	Configuration parameters

Returns: dict with run_number and metadata

run_info = workflow.submit({"batch_size": 64})
print(f"Started run #{run_info['run_number']}")

Get a specific run by number.

workflow.get_run(run_number: int) -> dict

Parameter	Type	Description
`run_number`	`int`	Run number

Returns: dict with run details

Update run status or metadata.

workflow.update_run(run_number: int, status: str = None, **kwargs) -> None

Parameter	Type	Description
`run_number`	`int`	Run number
`status`	`str`	New status

Delete the workflow.

workflow.delete() -> None

Clear cached data.

workflow.refresh() -> None

Create a new workflow.

Workflow.create(name: str, description: str = None) -> Workflow

Parameter	Type	Description
`name`	`str`	Workflow name
`description`	`str`	Optional description

Returns: Workflow

workflow = Workflow.create("my-pipeline", description="Data pipeline")

List all workflows in the workspace.

from mixtrain import list_workflows

workflows = list_workflows()
for wf in workflows:
    print(f"{wf.name}: {wf.description}")