Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1''' State file ''' 

2import cbor2 # type: ignore 

3from typing import Dict, Any, Optional 

4import gzip 

5import logging 

6import os 

7 

8 

9log = logging.getLogger(__name__) 

10VERSION: int = 1 

11 

12 

13class StateFile: 

14 #: The data 

15 d: Dict[str, Any] 

16 #: The filename we were loaded from, if any 

17 fname: Optional[str] 

18 

19 def __init__(self): 

20 self.d = { 

21 'version': VERSION, 

22 } 

23 self.fname = None 

24 

25 @staticmethod 

26 def from_file(fname: str) -> 'StateFile': 

27 ''' Load a state object from the given filename. If the file doesn't 

28 exist, just return a new object. ''' 

29 state = StateFile() 

30 state.fname = fname 

31 if not os.path.exists(fname): 31 ↛ 33line 31 didn't jump to line 33, because the condition on line 31 was never false

32 return state 

33 with gzip.open(fname, 'rb') as fd: 

34 state.d = cbor2.load(fd) 

35 if state.d['version'] > VERSION: 

36 log.warn( 

37 'Loaded state from %s with version %d, but the latest ' 

38 'version we know is %d. Bad things may happen.', 

39 fname, state.d['version'], VERSION) 

40 if state.d['version'] < VERSION: 

41 log.warn( 

42 'Need to update state format in %s from %d to %d, but nothing ' 

43 'to do that has been written yet.', 

44 fname, state.d['version'], VERSION) 

45 state.fname = fname 

46 return state 

47 

48 def to_file(self, fname: Optional[str] = None): 

49 ''' Write ourselves out to the given filename, overwriting anything 

50 that might already exist there. 

51 

52 - If no file is given and we don't know what file we were read from, do 

53 nothing. 

54 - If no file is given but we do know from where we were read, write out 

55 to that file. 

56 - If a file is given, write out to that regardless of where we were 

57 read (if anywhere). 

58 ''' 

59 if not fname and not self.fname: 

60 return 

61 if not fname: 

62 # log.debug('Writing state to %s', self.fname) 

63 with gzip.open(self.fname, 'wb') as fd: 

64 cbor2.dump(self.d, fd) 

65 return 

66 assert fname 

67 # log.debug('Writing state to %s', fname) 

68 with gzip.open(fname, 'wb') as fd: 

69 cbor2.dump(self.d, fd) 

70 

71 def set(self, key: str, val: Any, skip_write: bool = False): 

72 ''' Set ``key`` to ``val``, and write out this change to the state 

73 file, unless ``skip_write`` is set to ``True``. 

74 ''' 

75 # log.debug('Setting %s => %s', key, val) 

76 self.d[key] = val 

77 if not skip_write: 

78 self.to_file() 

79 

80 def get(self, key: str, default: Any = None) -> Any: 

81 ''' Get the value stored at ``key``, or the provided ``default`` value 

82 if there is no such key. By default, ``default`` is ``None``. ''' 

83 if key not in self.d: 

84 return default 

85 return self.d[key]