refactor!: remove progress parameter from Job wait methods

.pre-commit-config.yaml

@@ -48,7 +48,6 @@ repos:
         always_run: true
         args: [src, tests, examples]
         additional_dependencies:
-          - rich
           - typer
   - repo: https://github.com/scientific-python/cookie
     rev: 2024.04.23

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased](https://github.com/MaterialsPhysicsANU/pbspy/compare/v0.0.9...HEAD)
 
+### Removed
+
+- Remove `progress` parameter from the `wait[_all]` and `result[_all]` methods of `Job`
+  - These methods no longer print job IDs or progress
+- Remove `rich` dependency
+
 ## [0.0.9](https://github.com/MaterialsPhysicsANU/pbspy/releases/tag/v0.0.9) - 2026-05-18
 
 ### Added

examples/run.py

@@ -1,5 +1,4 @@
 import typer
-from rich import print
 
 from pbspy import Job, JobDescription, QueueLimits
 

pyproject.toml

@@ -19,9 +19,9 @@ classifiers = [
 ]
 license = {text = "MIT License"}
 dependencies = [
-  "rich~=13.8",
 ]
 
+
 [dependency-groups]
 dev = [
     "mypy>=1.14.1",

src/pbspy/__init__.py

@@ -11,8 +11,6 @@
 from dataclasses import dataclass, field
 from typing import Any, Self, TypeAlias
 
-from rich.progress import Progress, TextColumn, TimeElapsedColumn
-
 __all__ = ["JobDescription", "Job", "JobResult", "QueueLimits", "QueueLimitsMap", "gadi"]
 
 
@@ -66,24 +64,7 @@ def _try_get_exit_code(job_id: str) -> int | None:
         raise RuntimeError("Could not get exit code")
 
 
-def _pbs_wait_for_jobs(jobs: list[Job], progress: bool = True) -> None:
-    """
-    Waits for the PBS jobs to complete.
-
-    Args:
-        jobs (list[Job]): A list of Job objects representing the PBS jobs.
-        progress (bool): Whether or not to print the state of the job and status on exit.
-
-    Returns:
-        None
-    """
-    if progress:
-        _pbs_wait_for_jobs_with_progress(jobs)
-    else:
-        _pbs_wait_for_jobs_quiet(jobs)
-
-
-def _pbs_wait_for_jobs_quiet(jobs: list[Job]) -> None:
+def _pbs_wait_for_jobs(jobs: list[Job]) -> None:
     """
     Waits for the PBS jobs to complete.
 
@@ -111,56 +92,6 @@ def _pbs_wait_for_jobs_quiet(jobs: list[Job]) -> None:
             # raise RuntimeError(f"Failed to check job status: {process.stderr.decode('utf-8')}")
 
 
-def _pbs_wait_for_jobs_with_progress(jobs: list[Job]) -> None:
-    """
-    Waits for the PBS jobs to complete.
-
-    Args:
-        jobs (list[Job]): A list of Job objects representing the PBS jobs.
-
-    Returns:
-        None
-    """
-    last_check = time.time()
-    with Progress(
-        TimeElapsedColumn(),
-        TextColumn("[progress.description]{task.description}"),
-        auto_refresh=False,
-    ) as progress:
-        tasks = [progress.add_task(f"{job.job_id} {job.job_name} {job.description or ''}", total=1) for job in jobs]
-        task_done = [False] * len(jobs)
-        while not all(task_done):
-            time.sleep(1)
-            if time.time() - last_check > 60:
-                last_check = time.time()
-                for i, (job, task) in enumerate(zip(jobs, tasks, strict=False)):
-                    if task_done[i]:
-                        continue
-                    process = subprocess.run(
-                        ["qstat", job.job_id],
-                        capture_output=True,
-                    )
-                    if process.returncode != 0:
-                        output = process.stdout.decode("utf-8")
-                        if job.job_id not in output or "has finished" in process.stderr.decode("utf-8"):
-                            task_done[i] = True
-                            progress.advance(task)
-                            progress.update(task, visible=False)
-
-                            # Try and get the exit code
-                            exit_code = _try_get_exit_code(job.job_id)
-                            if exit_code is None:
-                                output_status = "[yellow]?"
-                            else:
-                                output_status = "[green]✓" if exit_code == 0 else "[red]✗"
-                            progress.console.print(
-                                f'{output_status} {job.job_id} {job.job_name} {job.description or ""}'
-                            )
-                    # else:
-                    # raise RuntimeError(f"Failed to check job status: {process.stderr.decode('utf-8')}")
-            progress.refresh()
-
-
 @dataclass
 class JobResult:
     """
@@ -196,14 +127,11 @@ class Job:
     description: str | None = None
     """A description of the job for progress updates. Unused by PBS."""
 
-    def wait(self, progress: bool = True) -> None:
+    def wait(self, **kwargs: dict[str, Any]) -> None:
         """
         Wait for the job to complete.
-
-        Args:
-            progress (bool): Whether or not to print the state of the job and status on exit.
         """
-        _pbs_wait_for_jobs([self], progress=progress)
+        _pbs_wait_for_jobs([self])
 
     def _result_no_wait(self) -> JobResult:
         """
@@ -254,35 +182,26 @@ def _result_no_wait(self) -> JobResult:
 
         return JobResult(exit_code=exit_code, output=output, error=error, stats=pbs_stats)
 
-    def result(self, progress: bool = True) -> JobResult:
+    def result(self, **kwargs: dict[str, Any]) -> JobResult:
         """
         Waits for the job to complete and returns the result.
-
-        Args:
-            progress (bool): Whether or not to print the state of the job and status on exit.
         """
-        self.wait(progress=progress)
+        self.wait()
         return self._result_no_wait()
 
     @staticmethod
-    def wait_all(jobs: list[Job], progress: bool = True) -> None:
+    def wait_all(jobs: list[Job], **kwargs: dict[str, Any]) -> None:
         """
         Waits for multiple jobs to complete.
-
-        Args:
-            progress (bool): Whether or not to print the state of the job and status on exit.
         """
-        _pbs_wait_for_jobs(jobs, progress=progress)
+        _pbs_wait_for_jobs(jobs)
 
     @staticmethod
-    def result_all(jobs: list[Job], progress: bool = True) -> list[JobResult]:
+    def result_all(jobs: list[Job], **kwargs: dict[str, Any]) -> list[JobResult]:
         """
         Waits for multiple jobs to complete and returns their results.
-
-        Args:
-            progress (bool): Whether or not to print the state of the job and status on exit.
         """
-        _pbs_wait_for_jobs(jobs, progress=progress)
+        _pbs_wait_for_jobs(jobs)
         return [job._result_no_wait() for job in jobs]