Coverage for /home/ubuntu/shekels/python/shekels/core/database.py: 100%

1from typing import List, Union # noqa: F401 

2from pathlib import Path # noqa: F401 

3 

4from copy import deepcopy 

5from functools import lru_cache 

6 

7import jsoncomment as jsonc 

8import numpy as np 

9import pandas as pd 

10 

11from shekels.core.config import Config 

12import shekels.core.data_tools as sdt 

13# ------------------------------------------------------------------------------ 

14 

15 

16class Database: 

17 ''' 

18 Database is a class for wrapping a mint transaction DataFrame with a simple 

19 CRUD-like API. API methods include: update, read and search. 

20 ''' 

21 @staticmethod 

22 def from_json(filepath): 

23 # type: (Union[str, Path]) -> Database 

24 ''' 

25 Constructs a Database instance from a given JSON filepath. 

26 

27 Args: 

28 filepath(Path or str): Path to JSON config file. 

29 

30 Returns: 

31 Database: Database instance. 

32 ''' 

33 with open(filepath) as f: 

34 config = jsonc.JsonComment().load(f) 

35 return Database(config) 

36 

37 def __init__(self, config): 

38 # type: (dict) -> None 

39 ''' 

40 Constructs a Database instance. 

41 

42 Args: 

43 config (dict): Configuration. 

44 ''' 

45 config = Config(config) 

46 config.validate() 

47 self._config = config.to_primitive() 

48 self._data = None # type: Union[None, pd.DataFrame] 

49 

50 @staticmethod 

51 def _to_records(data): 

52 # type: (pd.DataFrame) -> List[dict] 

53 ''' 

54 Converts given DataFrame to a list of JSONifiable dicts. 

55 

56 Args: 

57 data (DataFrame): Data. 

58 

59 Returns: 

60 list[dict]: Records. 

61 ''' 

62 data = data.copy() 

63 data.date = data.date.apply(lambda x: x.isoformat()) 

64 data = data.replace({np.nan: None}).to_dict(orient='records') 

65 return data 

66 

67 @property 

68 def config(self): 

69 # type: () -> dict 

70 ''' 

71 Returns a copy of this instance's configuration. 

72 

73 Returns: 

74 dict: Copy of config. 

75 ''' 

76 return deepcopy(self._config) 

77 

78 @property 

79 def data(self): 

80 # type: () -> Union[None, pd.DataFrame] 

81 ''' 

82 Returns a copy of this instance's data. 

83 

84 Returns: 

85 DataFrame: Copy of data. 

86 ''' 

87 if self._data is None: 

88 return None 

89 return self._data.copy() 

90 

91 def update(self): 

92 # type: () -> Database 

93 ''' 

94 Loads CSV found in config's data_path into self._data. 

95 

96 Returns: 

97 Database: self. 

98 ''' 

99 data = pd.read_csv(self._config['data_path'], index_col=None) 

100 self._data = sdt.conform( 

101 data, 

102 actions=self._config['conform'], 

103 columns=self._config['columns'], 

104 ) 

105 return self 

106 

107 @lru_cache(maxsize=1) 

108 def read(self): 

109 # type: () -> List[dict] 

110 ''' 

111 Returns data if update has been called. 

112 

113 Raises: 

114 RuntimeError: If update has not first been called. 

115 

116 Returns: 

117 list[dict]: Data as records. 

118 ''' 

119 if self._data is None: 

120 msg = 'Database not updated. Please call update.' 

121 raise RuntimeError(msg) 

122 return self._to_records(self._data) 

123 

124 @lru_cache() 

125 def search(self, query): 

126 # type: (str) -> List[dict] 

127 ''' 

128 Search data according to given SQL query. 

129 

130 Args: 

131 query (str): SQL query. Make sure to use "FROM data" in query. 

132 

133 Returns: 

134 DataFrame: Formatted data. 

135 ''' 

136 output = sdt.query_data(self._data, query) 

137 # pandasql coerces Timestamps to strings 

138 output.date = pd.DatetimeIndex(output.date) 

139 return self._to_records(output)