Coverage for src/sankey_cashflow/data_row.py: 89%

1from pandas._libs.tslibs import timestamps 

2 

3from .utils import is_empty, is_null 

4 

5 

6class DataRow: 

7 # static class - just a container for some related methods around single rows of expense data. 

8 # The columns we expect to see in the data 

9 fields = { 

10 "Date": { 

11 "required": True, 

12 "nullable": False, 

13 "type": timestamps.Timestamp, 

14 "force_type": False, 

15 "comment": "" 

16 }, 

17 "Category": { 

18 "required": True, 

19 "nullable": False, 

20 "type": str, 

21 "force_type": False, 

22 "comment": "" 

23 }, 

24 "Description": { 

25 "required": False, 

26 "nullable": True, 

27 "type": str, 

28 "force_type": False, 

29 "comment": "" 

30 }, 

31 "Tags": { 

32 "required": False, 

33 "nullable": True, 

34 "type": str, 

35 "force_type": False, 

36 "comment": "" 

37 }, 

38 "Comments": { 

39 "required": False, 

40 "nullable": True, 

41 "type": str, 

42 "force_type": False, 

43 "comment": "" 

44 }, 

45 "Source": { 

46 "required": False, 

47 "nullable": True, 

48 "type": str, 

49 "force_type": False, 

50 "comment": "" 

51 }, 

52 "Target": { 

53 "required": False, 

54 "nullable": True, 

55 "type": str, 

56 "force_type": False, 

57 "comment": "" 

58 }, 

59 "Type": { 

60 "required": False, 

61 "nullable": True, 

62 "type": str, 

63 "allowed_values": ["computed", "tag", ""], 

64 "force_type": False, 

65 "comment": "" 

66 }, 

67 "Distribution": { 

68 "required": False, 

69 "nullable": True, 

70 "type": int, 

71 "force_type": True, 

72 "comment": "Value in whole months to distribute the row amount over" 

73 }, 

74 "Amount": { 

75 "required": True, 

76 "nullable": False, 

77 "type": float, 

78 "force_type": True, 

79 "comment": "" 

80 }, 

81 "Sales Tax": { 

82 "required": False, 

83 "nullable": True, 

84 "type": float, 

85 "force_type": True, 

86 "comment": "" 

87 }, 

88 "Tips": { 

89 "required": False, 

90 "nullable": True, 

91 "type": float, 

92 "force_type": True, 

93 "comment": "" 

94 } 

95 } 

96 

97 @staticmethod 

98 def validate(drow, header_only=False, include_classifications=False): 

99 # Validate that data rows are correct 

100 this_fields = DataRow.fields.copy() # prevent mutation of the class fields 

101 if include_classifications: 

102 # Add classification to the fields 

103 this_fields["Classification"] = { 

104 "required": True, 

105 "nullable": False, 

106 "type": str, 

107 "force_type": False, 

108 "comment": "" 

109 } 

110 if len(drow) != len(this_fields): 

111 # import pdb; pdb.set_trace() 

112 raise Exception(f"Data rows should contain {len(DataRow.fields)} elements") 

113 if header_only: 

114 if drow != list(this_fields.keys()): 

115 return False, f"Data rows need to be in the form: {list(this_fields.keys())}" 

116 return True, None 

117 vkeys = list(this_fields.keys()) 

118 counter = 0 

119 while counter < len(drow): 

120 this_validator = this_fields[vkeys[counter]] # TODO: verify if this needs a copy 

121 this_value = drow[counter] 

122 counter += 1 

123 if is_null(this_value): # Also check for length = 0? 

124 if this_validator["nullable"]: 

125 pass 

126 else: 

127 raise Exception(f"Non-nullable field {vkeys[counter - 1]} was nulled in {drow}") 

128 else: 

129 if type(this_value) is not this_validator["type"]: 

130 if this_validator["force_type"]: 

131 try: 

132 this_value = this_validator["type"](this_value) 

133 except ValueError: 

134 raise Exception(f"Could not coerce \'{this_value}\' at idx {counter - 1} to \ 

135 {this_validator['type']} for this row: {drow}") 

136 if type(this_value) is not this_validator["type"]: 

137 raise Exception(f"Invalid type at idx {counter - 1} for {this_value} in {drow}") 

138 if this_validator.get("allowed_values") and this_value not in this_validator["allowed_values"]: 

139 raise Exception(f"Non-allowed value of {this_value} in {drow}") 

140 return drow 

141 

142 @staticmethod 

143 def create( 

144 date, 

145 category_name, 

146 source, target, 

147 amount, 

148 description="", 

149 sales_tax=0, 

150 tips=0, 

151 comment="", 

152 tags="", 

153 row_type="", 

154 distribution=0, 

155 classification="Uncategorized"): 

156 if is_null(amount): 

157 amount = 0 

158 else: 

159 try: 

160 amount = float(amount) 

161 except ValueError: 

162 amount = 0 

163 if is_null(tips): 

164 tips = 0 

165 else: 

166 try: 

167 tips = float(tips) 

168 except ValueError: 

169 tips = 0 

170 if is_null(distribution): 

171 distribution = 0 

172 else: 

173 try: 

174 distribution = int(distribution) 

175 except ValueError: 

176 distribution = 0 

177 if is_null(sales_tax): 

178 sales_tax = 0 

179 else: 

180 try: 

181 sales_tax = float(sales_tax) 

182 except ValueError: 

183 sales_tax = 0 

184 return DataRow.validate([ 

185 date, 

186 category_name, 

187 description, 

188 tags, 

189 comment, 

190 source, 

191 target, 

192 row_type, 

193 distribution, 

194 amount, 

195 sales_tax, 

196 tips, 

197 classification], False, True) 

198 

199 @staticmethod 

200 def tag_matches(row_tags, search_tags): 

201 # Tag logic 

202 this_exploded_tags = None 

203 if search_tags and not is_empty(search_tags) and not is_empty(row_tags): 

204 this_exploded_tags = [i.strip() for i in row_tags.split(',')] # TODO: Do this case insensitively? 

205 if this_exploded_tags: 

206 return [i for i in set(search_tags).intersection(set(this_exploded_tags))] 

207 return None