Coverage for model_workflow/utils/cache.py: 93%

1from model_workflow.utils.auxiliar import load_json, save_json 

2from model_workflow.utils.type_hints import * 

3 

4# The cache is used to store data to be reused between different runs 

5class Cache: 

6 def __init__ (self, cache_file : 'File'): 

7 # Save the previous cache file 

8 self.file = cache_file 

9 # Set an auxiliar file which is used to not overwrite directly the previous cache 

10 self._aux_file = cache_file.get_prefixed_file('.aux') 

11 # Set a dict to store the actual cached data 

12 self.data = {} 

13 # Load data from the cache file, if it already exists 

14 if self.file.exists: 

15 # Read the cache in disk 

16 previous_data = load_json(self.file.path) 

17 # Inherit every field 

18 for field_name, field_value in previous_data.items(): 

19 self.data[field_name] = field_value 

20 # Save the entry for the first time 

21 self.save() 

22 

23 # Read a value from the cache 

24 def retrieve (self, key : str, fallback = None): 

25 return self.data.get(key, fallback) 

26 

27 # Update the cache and save it to disk 

28 def update (self, key : str, value): 

29 self.data[key] = value 

30 self.save() 

31 

32 # Delete a value in the cache 

33 def delete (self, key : str): 

34 if key not in self.data: return 

35 del self.data[key] 

36 self.save() 

37 

38 # Reset the cache 

39 # This is called when some input files are modified 

40 def reset (self): 

41 self.data = {} 

42 self.save() 

43 

44 # Save the cache to disk, as a json file 

45 def save (self): 

46 # Write entries to disk 

47 # Write to the auxiliar file, thus not overwritting the original cache yet 

48 save_json(self.data, self._aux_file.path, indent = 4) 

49 # If the new cache is successfully written then replace the old one 

50 self._aux_file.rename_to(self.file)