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

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

2 

3from pathlib import Path 

4 

5from runrunner.base import Status 

6from runrunner.slurm import SlurmRun 

7 

8 

9def get_runs_from_file( 

10 path: Path, print_error: bool = False, filter: list[Status] = None 

11) -> list[SlurmRun]: 

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

13 

14 Args: 

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

16 print_error: Whether to print errors. 

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

18 returned. 

19 

20 Returns: 

21 List of all found SlumRun objects. 

22 """ 

23 runs = [] 

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

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

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

27 try: 

28 run_obj = SlurmRun.from_file(file) 

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

30 runs.append(run_obj) 

31 except Exception as ex: 

32 # Not a (correct) RunRunner JSON file 

33 if print_error: 

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

35 return runs