feat(cli): add lightning job logs

python/lightning_sdk/cli/job/__init__.py

@@ -8,11 +8,13 @@ def register_commands(group: click.Group) -> None:
     from lightning_sdk.cli.job.delete import delete_job
     from lightning_sdk.cli.job.inspect import inspect_job
     from lightning_sdk.cli.job.list import list_jobs
+    from lightning_sdk.cli.job.logs import logs_job
     from lightning_sdk.cli.job.run import run_job
     from lightning_sdk.cli.job.stop import stop_job
 
     group.add_command(run_job, name="run")
     group.add_command(list_jobs, name="list")
     group.add_command(inspect_job, name="inspect")
+    group.add_command(logs_job, name="logs")
     group.add_command(stop_job, name="stop")
     group.add_command(delete_job, name="delete")

python/lightning_sdk/cli/job/logs.py

@@ -0,0 +1,38 @@
+"""Job logs command."""
+
+from typing import Optional
+
+import rich_click as click
+
+from lightning_sdk.cli.legacy.job_and_mmt_action import _JobAndMMTAction
+from lightning_sdk.cli.utils.logging import LightningCommand
+from lightning_sdk.status import Status
+
+_TERMINAL = (Status.Completed, Status.Failed, Status.Stopped)
+
+
+@click.command("logs", cls=LightningCommand)
+@click.argument("name", required=False)
+@click.option(
+    "--teamspace",
+    default=None,
+    help=(
+        "the name of the teamspace the job lives in. "
+        "Should be specified as {teamspace_owner}/{teamspace_name} (e.g my-org/my-teamspace). "
+        "If not specified can be selected interactively."
+    ),
+)
+def logs_job(name: Optional[str] = None, teamspace: Optional[str] = None) -> None:
+    """Print the logs for a job.
+
+    Logs are available once the job reaches a terminal state (Completed, Failed or
+    Stopped). While the job is still pending or running this prints its current status
+    instead of erroring — re-run once it has finished.
+    """
+    job = _JobAndMMTAction().job(name=name, teamspace=teamspace)
+    if job.status not in _TERMINAL:
+        raise click.ClickException(
+            f"Job '{job.name}' is {job.status}; logs are only available once it reaches a "
+            "terminal state (Completed/Failed/Stopped). Re-run this command after it finishes."
+        )
+    click.echo(job.logs)

python/tests/cli/job/test_logs.py

@@ -0,0 +1,76 @@
+from unittest.mock import MagicMock
+
+from click.testing import CliRunner
+
+from lightning_sdk.status import Status
+from tests.cli.help import assert_help_contains, mock_command_logging
+
+
+@mock_command_logging
+def test_job_logs_help() -> None:
+    assert_help_contains(
+        "lightning job logs --help",
+        "Usage: lightning job logs",
+        "Print the logs for a job.",
+    )
+
+
+@mock_command_logging
+def test_jobs_logs_help() -> None:
+    assert_help_contains(
+        "lightning jobs logs --help",
+        "Usage: lightning jobs logs",
+        "Print the logs for a job.",
+    )
+
+
+@mock_command_logging
+def test_job_logs_help_shows_positional_name() -> None:
+    assert_help_contains("lightning job logs --help", "Usage: lightning job logs [OPTIONS] [NAME]")
+
+
+def _patch_action(monkeypatch, job: MagicMock, captured: dict) -> None:
+    class _FakeJobAndMMTAction:
+        def job(self, name=None, teamspace=None):
+            captured["name"] = name
+            captured["teamspace"] = teamspace
+            return job
+
+    monkeypatch.setattr("lightning_sdk.cli.job.logs._JobAndMMTAction", _FakeJobAndMMTAction)
+
+
+@mock_command_logging
+def test_job_logs_prints_logs_when_terminal(monkeypatch) -> None:
+    from lightning_sdk.cli.job.logs import logs_job
+
+    captured: dict = {}
+    job = MagicMock()
+    job.name = "my-job"
+    job.status = Status.Completed
+    job.logs = "hello from the job\n42"
+    _patch_action(monkeypatch, job, captured)
+
+    result = CliRunner().invoke(logs_job, ["my-job", "--teamspace", "org/teamspace"])
+
+    assert result.exit_code == 0
+    assert captured == {"name": "my-job", "teamspace": "org/teamspace"}
+    assert "hello from the job" in result.output
+    assert "42" in result.output
+
+
+@mock_command_logging
+def test_job_logs_errors_while_not_terminal(monkeypatch) -> None:
+    from lightning_sdk.cli.job.logs import logs_job
+
+    captured: dict = {}
+    job = MagicMock()
+    job.name = "my-job"
+    job.status = Status.Pending
+    _patch_action(monkeypatch, job, captured)
+
+    result = CliRunner().invoke(logs_job, ["my-job", "--teamspace", "org/teamspace"])
+
+    # errors cleanly (mentioning the status) rather than raising the raw SDK RuntimeError
+    assert result.exit_code != 0
+    assert "Pending" in result.output
+    assert not isinstance(result.exception, RuntimeError)