Coverage for gws-app/gws/plugin/csv_helper/__init__.py: 84%

1"""CSV helper. 

2 

3The ``csv`` helper writes CSV data with configurable formatting, for example 

4for the ALKIS export. ``writer`` creates a writer, which writes headers and 

5rows either into memory or directly into a binary stream. 

6 

7Values are formatted according to their type: 

8 

9- ``None`` becomes an empty quoted string, 

10- integers are written as they are; floats and decimals are formatted with 

11 the number formatter of the locale. Numbers are only quoted if 

12 ``quoteAll`` is set, 

13- dates, datetimes and times are formatted in the short format of the locale 

14 and quoted, 

15- other values are converted to strings and quoted. If ``formulaHack`` is 

16 set, digit-only strings are written as formulas (``="0123"``), so that 

17 spreadsheet programs keep leading zeros. 

18 

19The helper is created with default settings if it is not configured. 

20 

21Example:: 

22 

23 helpers+ { 

24 type "csv" 

25 format { 

26 delimiter ";" 

27 encoding "cp1252" 

28 rowDelimiter "CRLF" 

29 } 

30 } 

31 

32Usage in Python:: 

33 

34 helper = cast(gws.plugin.csv_helper.Object, root.app.helper('csv')) 

35 w = helper.writer(gws.lib.intl.locale('de_DE')) 

36 w.write_headers(['name', 'area']) 

37 w.write_row(['Parcel 1', 123.4]) 

38 data = w.to_bytes() 

39""" 

40 

41from typing import BinaryIO 

42 

43import decimal 

44import datetime 

45 

46import gws 

47import gws.lib.intl 

48 

49 

50class FormatConfig(gws.Config): 

51 """CSV format settings.""" 

52 

53 delimiter: str = ',' 

54 """Field delimiter.""" 

55 encoding: str = 'utf8' 

56 """Text encoding.""" 

57 formulaHack: bool = True 

58 """Write digit-only strings as formulas.""" 

59 quote: str = '"' 

60 """Quote character.""" 

61 quoteAll: bool = False 

62 """Quote all fields.""" 

63 rowDelimiter: str = 'LF' 

64 """Row delimiter.""" 

65 

66 

67@gws.ext.config.helper('csv') 

68class Config(gws.Config): 

69 """Format settings for CSV exports.""" 

70 

71 format: FormatConfig 

72 """CSV format settings.""" 

73 

74 

75class Format(gws.Data): 

76 """CSV format settings used by the writer.""" 

77 

78 delimiter: str 

79 """Field delimiter.""" 

80 encoding: str 

81 """Text encoding.""" 

82 formulaHack: bool 

83 """Write digit-only strings as formulas.""" 

84 quote: str 

85 """Quote character.""" 

86 quoteAll: bool 

87 """Quote all fields, including numbers.""" 

88 rowDelimiter: str 

89 """Row delimiter, with ``CR`` and ``LF`` replaced by the actual characters.""" 

90 

91 

92@gws.ext.object.helper('csv') 

93class Object(gws.Node): 

94 """CSV helper.""" 

95 

96 format: Format 

97 """Format settings.""" 

98 

99 def configure(self) -> None: 

100 self.format = Format( 

101 delimiter=self.cfg('format.delimiter', default=','), 

102 encoding=self.cfg('format.encoding', default='utf8'), 

103 formulaHack=self.cfg('format.formulaHack', default=True), 

104 quote=self.cfg('format.quote', default='"'), 

105 quoteAll=self.cfg('format.quoteAll', default=False), 

106 rowDelimiter=self.cfg('format.rowDelimiter', default='LF').replace('CR', '\r').replace('LF', '\n'), 

107 ) 

108 

109 def writer(self, locale: gws.Locale, stream_to: BinaryIO = None) -> '_Writer': 

110 """Create a CSV writer. 

111 

112 Args: 

113 locale: Locale for formatting numbers, dates and times. 

114 stream_to: Binary stream to write to. If ``None``, the data is kept in memory. 

115 

116 Returns: 

117 A new writer with the format settings of this helper. 

118 """ 

119 

120 return _Writer(self, locale, stream_to) 

121 

122 

123class _Writer: 

124 """CSV writer. 

125 

126 Writes headers and rows either directly into a binary stream or into 

127 memory. Data kept in memory is returned by ``to_str`` and ``to_bytes``. 

128 """ 

129 

130 def __init__(self, helper: 'Object', locale: gws.Locale, stream_to: BinaryIO = None) -> None: 

131 """Create a CSV writer. 

132 

133 Args: 

134 helper: The CSV helper with the format settings. 

135 locale: Locale for formatting numbers, dates and times. 

136 stream_to: Binary stream to write to. If ``None``, the data is kept in memory. 

137 """ 

138 self.helper: Object = helper 

139 self.format = self.helper.format 

140 self.stream_to = stream_to 

141 self.eol = self.format.rowDelimiter.encode(self.format.encoding) 

142 

143 self.headers = [] 

144 self.str_rows = [] 

145 self.str_headers = '' 

146 

147 f = gws.lib.intl.formatters(locale) 

148 self.dateFormatter = f[0] 

149 self.timeFormatter = f[1] 

150 self.numberFormatter = f[2] 

151 

152 def write_headers(self, headers: list[str]) -> '_Writer': 

153 """Write the header row. 

154 

155 The headers also define the column order for ``write_dict``. 

156 

157 Args: 

158 headers: Column names. 

159 

160 Returns: 

161 The writer itself, for chaining. 

162 """ 

163 

164 self.headers = headers 

165 self.str_headers = self.format.delimiter.join(self._quote(s) for s in headers) 

166 if self.stream_to: 

167 self.stream_to.write(self.str_headers.encode(self.format.encoding) + self.eol) 

168 return self 

169 

170 def write_row(self, row: list) -> '_Writer': 

171 """Write a data row. 

172 

173 Args: 

174 row: Values of the row. 

175 

176 Returns: 

177 The writer itself, for chaining. 

178 """ 

179 

180 s = self.format.delimiter.join(self._format(v) for v in row) 

181 if self.stream_to: 

182 self.stream_to.write(s.encode(self.format.encoding) + self.eol) 

183 else: 

184 self.str_rows.append(s) 

185 return self 

186 

187 def write_dict(self, d: dict) -> '_Writer': 

188 """Write a data row from a dict. 

189 

190 If no headers are written yet, the keys of the dict are written as 

191 headers first. Values are taken in the order of the headers, missing 

192 values are empty. 

193 

194 Args: 

195 d: Values by column name. 

196 

197 Returns: 

198 The writer itself, for chaining. 

199 """ 

200 

201 if not self.headers: 

202 self.write_headers(list(d.keys())) 

203 return self.write_row([d.get(h, '') for h in self.headers]) 

204 

205 def to_str(self) -> str: 

206 """Return the data kept in memory as a string. 

207 

208 Returns: 

209 The header row and the data rows, joined with the row delimiter. 

210 When writing into a stream, only the header row is kept in memory. 

211 """ 

212 

213 rows = [] 

214 if self.headers: 

215 rows.append(self.str_headers) 

216 rows.extend(self.str_rows) 

217 return self.format.rowDelimiter.join(rows) 

218 

219 def to_bytes(self, encoding: str = None) -> bytes: 

220 """Return the data kept in memory as bytes. 

221 

222 Characters that cannot be encoded are replaced. 

223 

224 Args: 

225 encoding: Text encoding. If ``None``, the format encoding is used. 

226 

227 Returns: 

228 The encoded CSV data. 

229 """ 

230 

231 return self.to_str().encode(encoding or self.format.encoding, errors='replace') 

232 

233 def _format(self, val) -> str: 

234 """Format a value according to its type.""" 

235 if val is None: 

236 return self._quote('') 

237 

238 if isinstance(val, (float, decimal.Decimal)): 

239 s = self.numberFormatter.decimal(val) 

240 return self._quote(s) if self.format.quoteAll else s 

241 

242 if isinstance(val, int): 

243 s = str(val) 

244 return self._quote(s) if self.format.quoteAll else s 

245 

246 if isinstance(val, (datetime.datetime, datetime.date)): 

247 s = self.dateFormatter.short(val) 

248 return self._quote(s) 

249 

250 if isinstance(val, datetime.time): 

251 s = self.timeFormatter.short(val) 

252 return self._quote(s) 

253 

254 val = gws.u.to_str(val) 

255 

256 if val and val.isdigit() and self.format.formulaHack: 

257 val = '=' + self._quote(val) 

258 

259 return self._quote(val) 

260 

261 def _quote(self, val) -> str: 

262 """Quote a value, doubling the quote characters in it.""" 

263 q = self.format.quote 

264 s = gws.u.to_str(val).replace(q, q + q) 

265 return q + s + q