Formatters

Formatters convert job parameters into executable commands for different execution environments.

Overview

WhatsNext provides three built-in formatters:

Formatter	Use Case
CLIFormatter	Direct command-line execution
SlurmFormatter	SLURM HPC cluster submission
RUNAIFormatter	Kubernetes/RUNAI submission

CLIFormatter

Executes jobs directly via subprocess.

Basic Usage

from whatsnext.api.client import CLIFormatter

formatter = CLIFormatter(
    executable="python",
    script="train.py",
    working_dir="/path/to/project"
)

# Job parameters: {"lr": 0.01, "epochs": 100}
# Generates: python train.py --lr 0.01 --epochs 100

Constructor Parameters

Parameter	Type	Default	Description
`executable`	str	`"python"`	The command to run
`script`	str	`None`	Script path to execute
`working_dir`	str	`None`	Working directory

How It Works

format(): Converts parameters to CLI arguments

formatter.format("train", {"lr": 0.01, "epochs": 100})
# Returns: ["python", "train.py", "--lr", "0.01", "--epochs", "100"]

execute(): Runs via subprocess

result = formatter.execute(["python", "train.py", "--lr", "0.01"])
print(result.returncode)  # 0 = success
print(result.stdout)      # Command output

SlurmFormatter

Submits jobs to SLURM HPC clusters using sbatch.

Basic Usage

from whatsnext.api.client import SlurmFormatter

formatter = SlurmFormatter(
    partition="gpu",
    time="04:00:00",
    nodes=1,
    cpus_per_task=8,
    mem="32G",
    gpus=4,
    script_template="python train.py --lr {lr} --epochs {epochs}"
)

Constructor Parameters

Parameter	Type	Default	Description
`partition`	str	`"default"`	SLURM partition
`time`	str	`"01:00:00"`	Wall time limit (HH:MM:SS)
`nodes`	int	`1`	Number of nodes
`ntasks`	int	`1`	Number of tasks
`cpus_per_task`	int	`1`	CPUs per task
`mem`	str	`"4G"`	Memory per node
`gpus`	int	`None`	Number of GPUs
`script_template`	str	`None`	Command template with placeholders

Script Template

Use Python format strings to inject parameters:

formatter = SlurmFormatter(
    partition="gpu",
    gpus=2,
    script_template="python train.py --lr {learning_rate} --batch {batch_size}"
)

# Job parameters: {"learning_rate": 0.001, "batch_size": 32}
# Generates sbatch command that runs:
#   python train.py --lr 0.001 --batch 32

Generated Command

formatter.format("my-job", {"lr": 0.01})
# Returns: ["sbatch", "--parsable", "--job-name", "my-job", "--wrap", "python --lr 0.01"]

RUNAIFormatter

Submits jobs to RUNAI on Kubernetes clusters.

Basic Usage

from whatsnext.api.client import RUNAIFormatter

formatter = RUNAIFormatter(
    project="ml-team",
    image="pytorch/pytorch:2.0-cuda11.8",
    gpu=4,
    cpu=16,
    memory="64Gi",
    working_dir="/workspace",
    environment={"WANDB_API_KEY": "xxx"}
)

Constructor Parameters

Parameter	Type	Default	Description
`project`	str	required	RUNAI project name
`image`	str	required	Docker image
`gpu`	int	`1`	Number of GPUs
`cpu`	int	`1`	Number of CPUs
`memory`	str	`"4Gi"`	Memory limit
`working_dir`	str	`None`	Container working directory
`environment`	dict	`None`	Environment variables

Environment Variables

Job parameters are passed as uppercase environment variables:

# Job parameters: {"learning_rate": 0.01, "batch_size": 32}
# Sets in container:
#   LEARNING_RATE=0.01
#   BATCH_SIZE=32

Generated Command

formatter.format("train-job", {"lr": 0.01})
# Returns: [
#   "runai", "submit", "train-job",
#   "--project", "ml-team",
#   "--image", "pytorch/pytorch:2.0-cuda11.8",
#   "--gpu", "4",
#   "--cpu", "16",
#   "--memory", "64Gi",
#   "-e", "LR=0.01"
# ]

Creating Custom Formatters

Extend the Formatter base class:

from whatsnext.api.client import Formatter
import subprocess
from typing import Any, Dict, List

class MyFormatter(Formatter):
    def __init__(self, **options):
        self.options = options

    def format(self, task: str, parameters: Dict[str, Any]) -> List[str]:
        """Convert parameters to a command."""
        cmd = ["my-tool", task]
        for key, value in parameters.items():
            cmd.extend([f"--{key}", str(value)])
        return cmd

    def execute(self, command: List[str]) -> subprocess.CompletedProcess:
        """Run the command."""
        return subprocess.run(
            command,
            capture_output=True,
            text=True
        )

API Reference

Formatter

Bases: ABC

Abstract base class for job formatters.

A Formatter converts job parameters into executable commands and handles their execution. Different formatters support different execution environments (CLI, SLURM, RUNAI, etc.).

execute `abstractmethod`

execute(command: List[str]) -> CompletedProcess

Execute the formatted command.

PARAMETER	DESCRIPTION
`command`	Command as a list of arguments. TYPE: `List[str]`

RETURNS	DESCRIPTION
`CompletedProcess`	CompletedProcess with return code, stdout, and stderr.

format `abstractmethod`

format(task: str, parameters: Dict[str, Any]) -> List[str]

Convert job parameters to an executable command.

PARAMETER	DESCRIPTION
`task`	The task name/template identifier. TYPE: `str`
`parameters`	Dictionary of job parameters. TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[str]`	Command as a list of arguments (for subprocess).

CLIFormatter

Bases: Formatter

Formatter for direct command-line execution.

Formats jobs as CLI commands and executes them via subprocess.

init

__init__(
    executable: str = "python",
    script: Optional[str] = None,
    working_dir: Optional[str] = None,
) -> None

Initialize a CLI formatter.

PARAMETER	DESCRIPTION
`executable`	The executable to run (e.g., "python", "bash"). TYPE: `str` DEFAULT: `'python'`
`script`	Optional script path to run. TYPE: `Optional[str]` DEFAULT: `None`
`working_dir`	Optional working directory for command execution. TYPE: `Optional[str]` DEFAULT: `None`

execute

execute(command: List[str]) -> CompletedProcess

Execute command via subprocess.

PARAMETER	DESCRIPTION
`command`	Command as list of arguments. TYPE: `List[str]`

RETURNS	DESCRIPTION
`CompletedProcess`	CompletedProcess with return code, stdout, and stderr.

format

format(task: str, parameters: Dict[str, Any]) -> List[str]

Format parameters as CLI arguments.

PARAMETER	DESCRIPTION
`task`	The task name (used to look up command templates). TYPE: `str`
`parameters`	Dictionary of parameters to pass as CLI args. TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[str]`	Command as list of arguments, e.g. ["python", "train.py", "--lr", "0.01"]

SlurmFormatter

Bases: Formatter

Formatter for SLURM batch job submission.

Generates sbatch scripts and submits them to SLURM.

init

__init__(
    partition: str = "default",
    time: str = "01:00:00",
    nodes: int = 1,
    ntasks: int = 1,
    cpus_per_task: int = 1,
    mem: str = "4G",
    gpus: Optional[int] = None,
    script_template: Optional[str] = None,
) -> None

Initialize a SLURM formatter.

PARAMETER	DESCRIPTION
`partition`	SLURM partition to submit to. TYPE: `str` DEFAULT: `'default'`
`time`	Maximum wall time (HH:MM:SS). TYPE: `str` DEFAULT: `'01:00:00'`
`nodes`	Number of nodes to request. TYPE: `int` DEFAULT: `1`
`ntasks`	Number of tasks. TYPE: `int` DEFAULT: `1`
`cpus_per_task`	CPUs per task. TYPE: `int` DEFAULT: `1`
`mem`	Memory per node. TYPE: `str` DEFAULT: `'4G'`
`gpus`	Number of GPUs (if any). TYPE: `Optional[int]` DEFAULT: `None`
`script_template`	Template for the job script body. TYPE: `Optional[str]` DEFAULT: `None`

execute

execute(command: List[str]) -> CompletedProcess

Submit to SLURM via sbatch.

PARAMETER	DESCRIPTION
`command`	sbatch command with arguments. TYPE: `List[str]`

RETURNS	DESCRIPTION
`CompletedProcess`	CompletedProcess with SLURM job ID in stdout on success.

format

format(task: str, parameters: Dict[str, Any]) -> List[str]

Format parameters as a SLURM submission command.

PARAMETER	DESCRIPTION
`task`	The task name. TYPE: `str`
`parameters`	Dictionary of job parameters. TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[str]`	sbatch command with inline script.

RUNAIFormatter

Bases: Formatter

Formatter for RUNAI job submission.

Submits jobs to RUNAI on Kubernetes clusters.

init

__init__(
    project: str,
    image: str,
    gpu: int = 1,
    cpu: int = 1,
    memory: str = "4Gi",
    working_dir: Optional[str] = None,
    environment: Optional[Dict[str, str]] = None,
) -> None

Initialize a RUNAI formatter.

PARAMETER	DESCRIPTION
`project`	RUNAI project name. TYPE: `str`
`image`	Docker image to use. TYPE: `str`
`gpu`	Number of GPUs to request. TYPE: `int` DEFAULT: `1`
`cpu`	Number of CPUs to request. TYPE: `int` DEFAULT: `1`
`memory`	Memory limit (e.g., "4Gi"). TYPE: `str` DEFAULT: `'4Gi'`
`working_dir`	Working directory inside container. TYPE: `Optional[str]` DEFAULT: `None`
`environment`	Environment variables to set. TYPE: `Optional[Dict[str, str]]` DEFAULT: `None`

execute

execute(command: List[str]) -> CompletedProcess

Submit to RUNAI.

PARAMETER	DESCRIPTION
`command`	runai submit command. TYPE: `List[str]`

RETURNS	DESCRIPTION
`CompletedProcess`	CompletedProcess with job submission result.

format

format(task: str, parameters: Dict[str, Any]) -> List[str]

Format parameters as a RUNAI submit command.

PARAMETER	DESCRIPTION
`task`	The task name (used as job name). TYPE: `str`
`parameters`	Dictionary of job parameters. TYPE: `Dict[str, Any]`

RETURNS	DESCRIPTION
`List[str]`	runai submit command.

Formatters

Overview

CLIFormatter

Basic Usage

Constructor Parameters

How It Works

SlurmFormatter

Basic Usage

Constructor Parameters

Script Template

Generated Command

RUNAIFormatter

Basic Usage

Constructor Parameters

Environment Variables

Generated Command

Creating Custom Formatters

API Reference

Formatter

execute abstractmethod

format abstractmethod

CLIFormatter

__init__

execute

format

SlurmFormatter

__init__

execute

format

RUNAIFormatter

__init__

execute

format

execute `abstractmethod`

format `abstractmethod`

init

init

init