"""Server sub-commands for the gyoza CLI."""
from __future__ import annotations
import json
import os
import typer
from gyoza.cli.errors import fail
app = typer.Typer(help="Manage the gyoza server.")
list_app = typer.Typer(help="List resources on the gyoza server.")
app.add_typer(list_app, name="list")
def _build_client() -> GyozaClient: # type: ignore[name-defined] # noqa: F821
"""Build a :class:`GyozaClient` from environment variables.
Returns
-------
GyozaClient
Connected client instance.
Raises
------
typer.Exit
If ``GYOZA_SERVER_URL`` is not set.
"""
from gyoza.client._client import GyozaClient # noqa: PLC0415
base_url = os.environ.get("GYOZA_SERVER_URL")
if not base_url:
fail("GYOZA_SERVER_URL environment variable is not set")
api_key = os.environ.get("GYOZA_API_KEY")
return GyozaClient(base_url=base_url, api_key=api_key)
[docs]
def start(
port: int = typer.Option(
5555,
"--port",
envvar="GYOZA_PORT",
help="Port the server listens on.",
),
worker_timeout: int = typer.Option(
60,
"--worker-timeout",
envvar="GYOZA_WORKER_TIMEOUT",
help="Seconds before a worker is considered inactive.",
),
scheduling_strategy: str = typer.Option(
"direct",
"--scheduling-strategy",
envvar="SCHEDULING_STRATEGY",
help="Scheduling strategy for allocating runs to workers.",
),
) -> None:
"""Start the gyoza server."""
os.environ["GYOZA_PORT"] = str(port)
os.environ["GYOZA_WORKER_TIMEOUT"] = str(worker_timeout)
os.environ["SCHEDULING_STRATEGY"] = scheduling_strategy
from gyoza.server import runner # noqa: PLC0415
runner.start()
def _format_table(rows: list[dict[str, str]], columns: list[tuple[str, str]]) -> str:
"""Format a list of dicts as a fixed-width table.
Parameters
----------
rows : list[dict[str, str]]
Row data keyed by column key.
columns : list[tuple[str, str]]
``(key, header)`` pairs defining column order and display names.
Returns
-------
str
Formatted table string.
"""
widths = {key: len(header) for key, header in columns}
for row in rows:
for key, _ in columns:
widths[key] = max(widths[key], len(row.get(key, "")))
header = " ".join(h.ljust(widths[k]) for k, h in columns)
separator = " ".join("-" * widths[k] for k, _ in columns)
lines = [header, separator]
for row in rows:
lines.append(" ".join(row.get(k, "").ljust(widths[k]) for k, _ in columns))
return "\n".join(lines)
def _truncate_ts(value: str) -> str:
"""Shorten an ISO timestamp to ``YYYY-MM-DD HH:MM``."""
return value.replace("T", " ")[:16] if value else ""
[docs]
def list_definitions(
as_json: bool = typer.Option(
False,
"--json",
help="Output full JSON instead of a table.",
),
) -> None:
"""List all OpDefinitions registered on the server."""
with _build_client() as client:
try:
definitions = client.list_definitions()
except Exception as exc: # noqa: BLE001
fail(f"Failed to fetch definitions: {exc}")
if not definitions:
typer.echo("No definitions found.")
raise typer.Exit()
if as_json:
typer.echo(json.dumps(definitions, indent=2))
return
table_columns: list[tuple[str, str]] = [
("id", "ID"),
("version", "VERSION"),
("image", "IMAGE"),
("created", "CREATED"),
("updated", "UPDATED"),
]
rows = [
{
"id": d.get("id", ""),
"version": d.get("version", ""),
"image": d.get("image", ""),
"created": _truncate_ts(d.get("createdAt", "")),
"updated": _truncate_ts(d.get("updatedAt", "")),
}
for d in definitions
]
typer.echo(_format_table(rows, table_columns))
app.command("start")(start)
list_app.command("definitions")(list_definitions)
