Coverage for sparkle/CLI/help/jobs.py: 79%

1"""File to help with RunRunner jobs.""" 

2from pathlib import Path 

3 

4from runrunner.base import Status 

5from runrunner.slurm import SlurmRun 

6 

7 

8def get_runs_from_file(path: Path, 

9 print_error: bool = False, 

10 filter: list[Status] = None) -> list[SlurmRun]: 

11 """Retrieve all run objects from file storage. 

12 

13 Args: 

14 path: Path object where to look recursively for the files. 

15 print_error: Whether to print errors. 

16 filter: If not None, only runs with the given statuses will be 

17 returned. 

18 

19 Returns: 

20 List of all found SlumRun objects. 

21 """ 

22 runs = [] 

23 for file in path.rglob("*.json"): 

24 # TODO: RunRunner should be adapted to have more general methods for runs 

25 # So this method can work for both local and slurm 

26 try: 

27 run_obj = SlurmRun.from_file(file) 

28 if filter is None or run_obj.status in filter: 

29 runs.append(run_obj) 

30 except Exception as ex: 

31 # Not a (correct) RunRunner JSON file 

32 if print_error: 

33 print(f"[WARNING] Could not load file: {file}. Exception: {ex}") 

34 return runs