Coverage for yaptide/batch/utils/utils.py: 31%

1import re 

2 

3 

4def sanitize_string(target_str: str, allowed_chars: str = r'[\w\-.,=/:]+') -> str: 

5 """Function clearing unaccepted signs""" 

6 return re.sub(f'[^\\s{allowed_chars}]', '', target_str) 

7 

8 

9def extract_sbatch_header(payload_dict: dict, target_key: str) -> str: 

10 """Function extracting header for slurm script""" 

11 return (sanitize_string(payload_dict["batch_options"][target_key], r'[\w\-.,=/:#]+') 

12 if "batch_options" in payload_dict 

13 and target_key in payload_dict["batch_options"] 

14 else "") 

15 

16 

17def convert_dict_to_sbatch_options(payload_dict: dict, target_key: str) -> str: 

18 """Function converting dict to sbatch command line options""" 

19 options_dict = { 

20 "time": "00:59:59" 

21 } 

22 if "batch_options" in payload_dict and target_key in payload_dict["batch_options"]: 

23 options_dict.update(payload_dict["batch_options"][target_key]) 

24 opt_list = [] 

25 for key, val in options_dict.items(): 

26 opt_list.append(f"--{sanitize_string(key)}={sanitize_string(val)}") 

27 return " ".join(opt_list)