feat(CLI): add tg batches commands for the batch API

src/together/lib/cli/__init__.py

@@ -31,6 +31,7 @@
     EVALS_HELP_EXAMPLES,
     FILES_HELP_EXAMPLES,
     MODELS_HELP_EXAMPLES,
+    BATCHES_HELP_EXAMPLES,
     JIG_LOGS_HELP_EXAMPLES,
     JIG_PUSH_HELP_EXAMPLES,
     ENDPOINTS_HELP_EXAMPLES,
@@ -47,9 +48,11 @@
     FILES_UPLOAD_HELP_EXAMPLES,
     BETA_CLUSTERS_HELP_EXAMPLES,
     MODELS_UPLOAD_HELP_EXAMPLES,
+    BATCHES_SUBMIT_HELP_EXAMPLES,
     BETA_ENDPOINTS_HELP_EXAMPLES,
     JIG_JOB_STATUS_HELP_EXAMPLES,
     JIG_SECRETS_SET_HELP_EXAMPLES,
+    BATCHES_DOWNLOAD_HELP_EXAMPLES,
     ENDPOINTS_CREATE_HELP_EXAMPLES,
     ENDPOINTS_UPDATE_HELP_EXAMPLES,
     BETA_ENDPOINTS_AB_HELP_EXAMPLES,
@@ -540,6 +543,23 @@ async def run_command() -> None:
 evals_app.command((f"{_CLI}.evals.retrieve:retrieve"), alias="get", help="Get eval job details")
 evals_app.command((f"{_CLI}.evals.status:status"), help="Get an eval job's status")
 
+## Batches API commands
+batches_app = app.command(App(name="batches", help="Submit and manage batch jobs", help_epilogue=BATCHES_HELP_EXAMPLES))
+batches_app.command(
+    (f"{_CLI}.batches.submit:submit"),
+    help="Submit a new batch job",
+    help_epilogue=BATCHES_SUBMIT_HELP_EXAMPLES,
+    sort_key=1,
+)
+batches_app.command((f"{_CLI}.batches.list:list"), alias="ls", help="List batch jobs")
+batches_app.command((f"{_CLI}.batches.retrieve:retrieve"), alias="get", help="Get batch job details")
+batches_app.command(
+    (f"{_CLI}.batches.download:download"),
+    help="Download batch job output (and error) files",
+    help_epilogue=BATCHES_DOWNLOAD_HELP_EXAMPLES,
+)
+batches_app.command((f"{_CLI}.batches.cancel:cancel"), help="Cancel a batch job")
+
 ## Telemetry API commands
 telemetry_app = app.command(App(name="telemetry", help="Configure CLI telemetry"))
 telemetry_app.command((f"{_CLI}.telemetry.status:status"), help="Show telemetry status")

src/together/lib/cli/api/batches/_utils.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import Literal
+from typing_extensions import TypeAlias
+
+from rich.markup import escape as escape_rich_markup
+
+from together.lib.utils.tools import format_datetime
+from together.types.batch_job import BatchJob
+from together.lib.cli.utils._console import console
+
+BatchApiType: TypeAlias = Literal["chat.completions", "audio.transcriptions", "audio.translations"]
+BatchEndpoint: TypeAlias = Literal["/v1/chat/completions", "/v1/audio/transcriptions", "/v1/audio/translations"]
+
+API_TO_ENDPOINT: dict[BatchApiType, BatchEndpoint] = {
+    "chat.completions": "/v1/chat/completions",
+    "audio.transcriptions": "/v1/audio/transcriptions",
+    "audio.translations": "/v1/audio/translations",
+}
+
+ENDPOINT_TO_API: dict[str, BatchApiType] = {endpoint: api for api, endpoint in API_TO_ENDPOINT.items()}
+
+STATUS_COLORS = {
+    "VALIDATING": "yellow",
+    "IN_PROGRESS": "yellow",
+    "COMPLETED": "green",
+    "FAILED": "red",
+    "EXPIRED": "red",
+    "CANCELLED": "red",
+}
+
+_INCOMPLETE_STATUSES = frozenset({"VALIDATING", "IN_PROGRESS"})
+_DOWNLOADABLE_STATUSES = frozenset({"COMPLETED", "FAILED", "EXPIRED", "CANCELLED"})
+_PROGRESS_BAR_WIDTH = 20
+
+
+def format_endpoint(endpoint: str | None) -> str:
+    if not endpoint:
+        return ""
+    return ENDPOINT_TO_API.get(endpoint, endpoint)
+
+
+def format_status(status: str | None) -> str:
+    if not status:
+        return ""
+    color = STATUS_COLORS.get(status, "white")
+    return f"[bold {color}]{status.capitalize()}[/bold {color}]"
+
+
+def format_progress(progress: float) -> str:
+    pct = max(0.0, min(100.0, progress))
+    filled = round((pct / 100.0) * _PROGRESS_BAR_WIDTH)
+    bar = "█" * filled + "░" * (_PROGRESS_BAR_WIDTH - filled)
+    return f"[yellow]{bar}[/yellow] [bold]{pct:g}%[/bold]"
+
+
+def print_batch_detail(job: BatchJob) -> None:
+    """Print a curated human-readable view of a batch job."""
+    console.print("Batch job details:")
+    if job.created_at:
+        console.print(f" - Created at {format_datetime(job.created_at)}")
+    if job.status == "COMPLETED" and job.completed_at:
+        console.print(f" - Completed at {format_datetime(job.completed_at)}")
+
+    api = format_endpoint(job.endpoint)
+    if api:
+        console.print(f" - {api}")
+
+    if job.x_model_id:
+        console.print(f" - {job.x_model_id}")
+
+    if job.output_file_id:
+        console.print(f" - Output file ID {job.output_file_id}")
+
+    if job.error_file_id or job.error:
+        console.print(" - [red]An error occurred[/red]")
+        if job.error_file_id:
+            console.print(f" - Error file ID {job.error_file_id}")
+        if job.error:
+            console.print(f"   {escape_rich_markup(job.error)}")
+
+    if job.status in _INCOMPLETE_STATUSES and job.progress is not None:
+        console.print(f"{format_progress(job.progress)} {format_status(job.status)}")
+
+    if job.status in _DOWNLOADABLE_STATUSES and job.id and (job.output_file_id or job.error_file_id):
+        console.print("\nDownload results with:")
+        console.print(f"[dim]-[/dim] [primary]tg batches download {job.id} --output ./out[/primary]")

src/together/lib/cli/api/batches/cancel.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from cyclopts import Parameter
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.batches._utils import print_batch_detail
+
+
+async def cancel(
+    batch_id: Annotated[str, Parameter(help="The ID of the batch job to cancel")],
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Cancel a batch job."""
+    response = await show_loading_status("Cancelling batch job...", config.client.batches.cancel(batch_id))
+    if config.json:
+        console.print_json(openapi_dumps(response).decode("utf-8"))
+        return
+
+    console.print("[green]√[/green] Cancelled batch job")
+    print_batch_detail(response)

src/together/lib/cli/api/batches/download.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import sys
+from typing import NoReturn, Optional, Annotated
+from pathlib import Path
+
+from cyclopts import Parameter, validators
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console, error_console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.files.retrieve_content import (
+    is_directory_output,
+    download_file_content,
+    stream_file_content_to_stdout,
+)
+
+_TERMINAL_STATUSES = frozenset({"COMPLETED", "FAILED", "EXPIRED", "CANCELLED"})
+
+
+def _error_output_path(output: Path) -> Path:
+    """Where to write the error file when *output* is a concrete file path."""
+    suffix = output.suffix or ".jsonl"
+    return output.with_name(f"{output.stem}.errors{suffix}")
+
+
+def _fail(*, json_mode: bool, error: str, rich_message: str) -> NoReturn:
+    if json_mode:
+        console.print_json(openapi_dumps({"error": error}).decode("utf-8"))
+    else:
+        console.print(rich_message)
+    sys.exit(1)
+
+
+async def download(
+    id: Annotated[str, Parameter(help="The ID of the batch job")],
+    output: Annotated[
+        Optional[Path],
+        Parameter(
+            name=["--output", "-o"],
+            help="File or directory to save batch result files to; omit to print output to stdout",
+            validator=validators.Path(file_okay=True, dir_okay=True),
+        ),
+    ] = None,
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Download output (and error) files for a batch job."""
+    job = await show_loading_status("Retrieving batch job...", config.client.batches.retrieve(id))
+    status = job.status or ""
+
+    if status not in _TERMINAL_STATUSES:
+        _fail(
+            json_mode=config.json,
+            error=(
+                f"Batch job is not ready to download yet (status: {status or 'unknown'}). "
+                f"Check progress with tg batches get {id}."
+            ),
+            rich_message=(
+                f"[red]Batch job is not ready to download yet[/red] "
+                f"(status: {status or 'unknown'}). "
+                f"Check progress with [primary]tg batches get {id}[/primary]."
+            ),
+        )
+
+    if not job.output_file_id and not job.error_file_id:
+        _fail(
+            json_mode=config.json,
+            error=f"Batch job has no output or error files to download (status: {status}).",
+            rich_message=f"[red]Batch job has no output or error files to download[/red] (status: {status}).",
+        )
+
+    if output is not None:
+        saved: list[dict[str, str]] = []
+        directory_output = is_directory_output(output)
+        error_file_id = job.error_file_id
+
+        if job.output_file_id:
+            out_path = await download_file_content(
+                config.client,
+                job.output_file_id,
+                output=output,
+                loading_message="Downloading batch output...",
+            )
+            assert isinstance(out_path, Path)
+            saved.append({"kind": "output", "id": job.output_file_id, "path": str(out_path)})
+        elif error_file_id and not directory_output:
+            # No output file — write the error file to the exact path the user asked for.
+            err_path = await download_file_content(
+                config.client,
+                error_file_id,
+                output=output,
+                loading_message="Downloading batch errors...",
+            )
+            assert isinstance(err_path, Path)
+            saved.append({"kind": "error", "id": error_file_id, "path": str(err_path)})
+            error_file_id = None
+
+        if error_file_id:
+            err_dest = output if directory_output else _error_output_path(output)
+            err_path = await download_file_content(
+                config.client,
+                error_file_id,
+                output=err_dest,
+                loading_message="Downloading batch errors...",
+            )
+            assert isinstance(err_path, Path)
+            saved.append({"kind": "error", "id": error_file_id, "path": str(err_path)})
+
+        if config.json:
+            console.print_json(openapi_dumps({"batch_id": id, "files": saved}).decode("utf-8"))
+            return
+
+        for item in saved:
+            label = "Output" if item["kind"] == "output" else "Errors"
+            console.print(f"[green]√[/green] {label} saved to [blue]{item['path']}[/blue]")
+        return
+
+    output_file_id = job.output_file_id
+    if not output_file_id:
+        _fail(
+            json_mode=config.json,
+            error="Batch job has no output file. Use --output to download the error file instead.",
+            rich_message=(
+                "[red]Batch job has no output file[/red]. "
+                "Use [primary]--output[/primary] to download the error file instead."
+            ),
+        )
+
+    if config.json:
+        payload: dict[str, str] = {
+            "batch_id": id,
+            "output_file_id": output_file_id,
+        }
+        if job.error_file_id:
+            payload["error_file_id"] = job.error_file_id
+        console.print_json(openapi_dumps(payload).decode("utf-8"))
+        return
+
+    await stream_file_content_to_stdout(config.client, output_file_id)
+    if job.error_file_id:
+        error_console.print(
+            f"[dim]Error file also available: tg batches download {id} --output ./out[/dim]",
+        )

src/together/lib/cli/api/batches/list.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from together._utils._json import openapi_dumps
+from together.lib.utils.tools import format_datetime
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.list import ListTable
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.batches._utils import STATUS_COLORS, format_endpoint
+from together.lib.cli.utils._mock_pagination import AfterParameter, mock_pagination
+
+
+async def list(
+    after: AfterParameter = None,
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """List batch jobs."""
+    response = await show_loading_status("Loading batch jobs...", config.client.batches.list())
+    jobs = response or []
+
+    epoch_start = datetime.fromtimestamp(0, tz=timezone.utc)
+    jobs.sort(key=lambda x: x.created_at or epoch_start, reverse=True)
+
+    jobs_to_display, next_cursor = mock_pagination(jobs, cursor_field="id", cursor=after)
+
+    if config.json:
+        console.print_json(openapi_dumps(jobs_to_display).decode("utf-8"))
+        return
+
+    table = ListTable(
+        empty_message="You don't have any batch jobs yet. To submit your first batch run:\n  [dim]-[/dim] [primary]tg batches submit[/primary]"
+    )
+    table.add_primary_column("ID")
+    table.add_column("API")
+    table.add_column("Model")
+    table.add_column("Status")
+    table.add_column("Created At")
+
+    for job in jobs_to_display:
+        status = str(job.status) if job.status is not None else ""
+        status_color = STATUS_COLORS.get(status, "white")
+        if job.status == "IN_PROGRESS" and job.progress is not None:
+            status = f"{status}: {job.progress:g}%"
+
+        table.add_row(
+            job.id or "",
+            format_endpoint(job.endpoint),
+            job.x_model_id or "",
+            f"[{status_color}]{status}[/{status_color}]",
+            format_datetime(job.created_at) if job.created_at else "",
+        )
+    console.print(table)
+    if next_cursor:
+        console.print("\n[blue dim]To display the next page, run:[/blue dim]")
+        console.print(f"  [dim]-[/dim] [white]tg batches list --after {next_cursor}[/white]")

src/together/lib/cli/api/batches/retrieve.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from cyclopts import Parameter
+
+from together._utils._json import openapi_dumps
+from together.lib.cli.utils.config import CLIConfigParameter
+from together.lib.cli.utils._console import console
+from together.lib.cli.components.loader import show_loading_status
+from together.lib.cli.api.batches._utils import print_batch_detail
+
+
+async def retrieve(
+    batch_id: Annotated[str, Parameter(help="The ID of the batch job")],
+    *,
+    config: CLIConfigParameter,
+) -> None:
+    """Get details of a batch job."""
+    response = await show_loading_status("Retrieving batch job...", config.client.batches.retrieve(batch_id))
+    if config.json:
+        console.print_json(openapi_dumps(response).decode("utf-8"))
+        return
+
+    print_batch_detail(response)