Coverage for gws-app/gws/config/parser.py: 17%

1"""Configuration parser. 

2 

3Convert configuration files (in different formats) or row config dicts 

4into ``gws.Config`` objects by validating them against the specs. 

5""" 

6 

7from typing import Optional, cast 

8 

9import os 

10import yaml 

11 

12import gws 

13import gws.lib.jsonx 

14import gws.lib.osx 

15import gws.lib.datetimex 

16import gws.lib.dynimport 

17import gws.lib.vendor.jump 

18import gws.lib.vendor.slon 

19import gws.spec.runtime 

20 

21CONFIG_PATH_PATTERN = r'\.(py|json|yaml|yml|cx)$' 

22 

23 

24def parse_from_path(path: str, as_type: str, ctx: gws.ConfigContext) -> Optional[gws.Config]: 

25 """Parse a configuration from a path. 

26 

27 Args: 

28 path: Path to the configuration file. 

29 as_type: Type of the configuration (e.g., 'gws.base.application.core.Config'). 

30 ctx: Configuration context. 

31 """ 

32 

33 pp = _Parser(ctx) 

34 val = pp.read_from_path(path) 

35 d = pp.ensure_dict(val, path) 

36 return pp.parse_dict(d, path, as_type) if d else None 

37 

38 

39def parse_dict(dct: dict | gws.Data, path: str, as_type: str, ctx: gws.ConfigContext) -> Optional[gws.Config]: 

40 """Parse a configuration given as python dict. 

41 

42 Args: 

43 dct: Dictionary containing the configuration. 

44 path: Path to the configuration file (for error reporting). 

45 as_type: Type of the configuration. 

46 ctx: Configuration context. 

47 """ 

48 

49 pp = _Parser(ctx) 

50 d = pp.ensure_dict(dct, path) 

51 return pp.parse_dict(d, path, as_type) if d else None 

52 

53 

54def parse_app_from_path(path: str, ctx: gws.ConfigContext) -> Optional[gws.Config]: 

55 """Parse application configuration from a path. 

56 

57 Args: 

58 path: Path to the application configuration file. 

59 ctx: Configuration context. 

60 """ 

61 

62 pp = _Parser(ctx) 

63 val = pp.read_from_path(path) 

64 d = pp.ensure_dict(val, path) 

65 return _parse_app_dict(d, path, pp) if d else None 

66 

67 

68def parse_app_dict(dct: dict | gws.Data, path: str, ctx: gws.ConfigContext) -> Optional[gws.Config]: 

69 """Parse application configuration given as python dict. 

70 

71 Args: 

72 dct: Dictionary containing the application configuration. 

73 path: Path to the configuration file (for error reporting). 

74 ctx: Configuration context. 

75 """ 

76 

77 pp = _Parser(ctx) 

78 d = pp.ensure_dict(dct, path) 

79 return _parse_app_dict(d, path, pp) if d else None 

80 

81 

82def read_from_path(path: str, ctx: gws.ConfigContext) -> Optional[dict]: 

83 """Read a configuration file from a path, parse config formats. 

84 

85 Args: 

86 path: Path to the configuration file. 

87 ctx: Configuration context. 

88 """ 

89 pp = _Parser(ctx) 

90 val = pp.read_from_path(path) 

91 d = pp.ensure_dict(val, path) 

92 return d 

93 

94 

95## 

96 

97 

98def _parse_app_dict(dct: dict, path, pp: '_Parser'): 

99 dct = gws.u.to_dict(dct) 

100 if not isinstance(dct, dict): 

101 _register_error(pp.ctx, f'app config must be a dict', path=path) 

102 return 

103 

104 # the timezone must be set before everything else 

105 tz = dct.get('server', {}).get('timeZone', '') 

106 if tz: 

107 gws.lib.datetimex.set_local_time_zone(tz) 

108 gws.log.info(f'local time zone="{gws.lib.datetimex.time_zone()}"') 

109 

110 # remove 'projects' from the config, parse them later on 

111 inline_projects = dct.pop('projects', []) 

112 

113 app_cfg = pp.parse_dict(dct, path, as_type='gws.base.application.core.Config') 

114 if not app_cfg: 

115 return 

116 

117 projects = [] 

118 for dcts in inline_projects: 

119 projects.extend(_parse_projects(dcts, path, pp)) 

120 

121 project_paths = list(app_cfg.get('projectPaths') or []) 

122 project_dirs = list(app_cfg.get('projectDirs') or []) 

123 

124 all_project_paths = list(project_paths) 

125 for dirname in project_dirs: 

126 all_project_paths.extend(gws.lib.osx.find_files(dirname, CONFIG_PATH_PATTERN, deep=True)) 

127 

128 for pth in sorted(set(all_project_paths)): 

129 projects.extend(_parse_projects_from_path(pth, pp)) 

130 

131 app_cfg.set('projectPaths', project_paths) 

132 app_cfg.set('projectDirs', project_dirs) 

133 app_cfg.set('projects', projects) 

134 

135 gws.log.if_debug(_save_debug, app_cfg, path, '.parsed.json') 

136 return app_cfg 

137 

138 

139def _parse_projects_from_path(path, pp: '_Parser'): 

140 cfg_list = pp.read_from_path(path) 

141 if not cfg_list: 

142 return [] 

143 return _parse_projects(cfg_list, path, pp) 

144 

145 

146def _parse_projects(cfg_list, path, pp: '_Parser'): 

147 ps = [] 

148 

149 for c in _as_flat_list(cfg_list): 

150 d = pp.ensure_dict(c, path) 

151 if not d: 

152 continue 

153 prj_cfg = pp.parse_dict(d, path, 'gws.ext.config.project') 

154 if prj_cfg: 

155 ps.append(prj_cfg) 

156 

157 return ps 

158 

159 

160## 

161 

162 

163class _Parser: 

164 def __init__(self, ctx: gws.ConfigContext): 

165 self.ctx = ctx 

166 self.ctx.errors = ctx.errors or [] 

167 self.ctx.paths = ctx.paths or set() 

168 self.ctx.readOptions = ctx.readOptions or set() 

169 self.ctx.readOptions.add(gws.SpecReadOption.verboseErrors) 

170 

171 def ensure_dict(self, val, path): 

172 if val is None: 

173 return 

174 d = _to_plain(val) 

175 if not isinstance(d, dict): 

176 _register_error(self.ctx, f'unsupported configuration type {type(val)!r}', path=path) 

177 return 

178 return d 

179 

180 def parse_dict(self, dct: dict, path: str, as_type: str) -> Optional[gws.Config]: 

181 if not isinstance(dct, dict): 

182 _register_error(self.ctx, 'unsupported configuration', path=path) 

183 return 

184 if path: 

185 _register_path(self.ctx, path) 

186 try: 

187 cfg = self.ctx.specs.read( 

188 dct, 

189 as_type, 

190 path=path, 

191 options=self.ctx.readOptions, 

192 ) 

193 return cast(gws.Config, cfg) 

194 except gws.spec.runtime.ReadError as exc: 

195 message, _, cei = exc.args 

196 _register_error(self.ctx, f'parse error: {message}', cei=cei) 

197 

198 def read_from_path(self, path: str): 

199 if not os.path.isfile(path): 

200 _register_error(self.ctx, f'file not found', path=path) 

201 return 

202 

203 _register_path(self.ctx, path) 

204 r = self.read2(path) 

205 

206 if r: 

207 r = _to_plain(r) 

208 gws.log.if_debug(_save_debug, r, path, '.src.json') 

209 return r 

210 

211 def read2(self, path: str): 

212 if path.endswith('.py'): 

213 return self.read_py(path) 

214 if path.endswith('.json'): 

215 return self.read_json(path) 

216 if path.endswith('.yml') or path.endswith('.yaml'): 

217 return self.read_yaml(path) 

218 if path.endswith('.cx'): 

219 return self.read_cx(path) 

220 

221 _register_error(self.ctx, 'unsupported configuration', path=path) 

222 

223 def read_py(self, path: str): 

224 try: 

225 fn = gws.lib.dynimport.load_file(path).get('main') 

226 if not fn: 

227 _register_error(self.ctx, f'no "main" function found', path=path) 

228 return 

229 return fn(self.ctx) 

230 except Exception as exc: 

231 gws.log.exception() 

232 _register_error(self.ctx, f'python error: {exc}', path=path) 

233 

234 def read_json(self, path: str): 

235 try: 

236 return gws.lib.jsonx.from_path(path) 

237 except Exception as exc: 

238 _register_error(self.ctx, f'json error: {exc}', path=path) 

239 

240 def read_yaml(self, path: str): 

241 try: 

242 with open(path, encoding='utf8') as fp: 

243 return yaml.safe_load(fp) 

244 except Exception as exc: 

245 _register_error(self.ctx, f'yaml error: {exc}', path=path) 

246 

247 def read_cx(self, path: str): 

248 err_cnt = [0] 

249 

250 def _error_handler(exc, path, line, env): 

251 _register_syntax_error(self.ctx, path, gws.u.read_file(path), message=repr(exc), line=line) 

252 err_cnt[0] += 1 

253 return True 

254 

255 def _loader(cur_path, load_path): 

256 if not os.path.isabs(load_path): 

257 load_path = os.path.abspath(os.path.dirname(cur_path) + '/' + load_path) 

258 _register_path(self.ctx, load_path) 

259 return gws.u.read_file(load_path), load_path 

260 

261 try: 

262 tpl = gws.lib.vendor.jump.compile_path(path, loader=_loader) 

263 except gws.lib.vendor.jump.CompileError as exc: 

264 _register_syntax_error(self.ctx, path, gws.u.read_file(exc.path), message=exc.message, line=exc.line) 

265 return 

266 

267 args = args = { 

268 'true': True, 

269 'false': False, 

270 'ctx': self.ctx, 

271 'gws': gws, 

272 } 

273 

274 slon = gws.lib.vendor.jump.call(tpl, args, error=_error_handler) 

275 if err_cnt[0] > 0: 

276 return 

277 

278 gws.log.if_debug(_save_debug, slon, path, '.src.slon') 

279 

280 try: 

281 return gws.lib.vendor.slon.loads(slon, as_object=True) 

282 except gws.lib.vendor.slon.SlonError as exc: 

283 _register_syntax_error(self.ctx, path, slon, message=exc.args[0], line=exc.args[2]) 

284 

285 

286## 

287 

288 

289def _register_path(ctx, path): 

290 ctx.paths.add(path) 

291 

292 

293def _register_error(ctx: gws.ConfigContext, message: str, **kwargs): 

294 cei = kwargs.pop('cei', None) or gws.ConfigErrorInfo() 

295 cei.message = message 

296 cei.update(kwargs) 

297 ctx.errors.append(cei) 

298 

299 

300def _register_syntax_error(ctx, path, src, message, line, context=10, cause=None): 

301 cei = gws.ConfigErrorInfo( 

302 path=path, 

303 line=line, 

304 message=f'syntax error: {message}', 

305 contextLines=[], 

306 cause=cause, 

307 ) 

308 

309 for n, ln in enumerate(src.splitlines(), 1): 

310 if n < line - context: 

311 continue 

312 if n > line + context: 

313 break 

314 ln = f'{n}: {ln}' 

315 if n == line: 

316 ln = f'>>> {ln}' 

317 cei.contextLines.append(ln) 

318 

319 ctx.errors.append(cei) 

320 

321 

322def _save_debug(src, src_path, ext): 

323 if ext.endswith('.json') and not isinstance(src, str): 

324 src = gws.lib.jsonx.to_pretty_string(src) 

325 path = gws.u.write_file(f'{gws.c.CONFIG_DIR}/{gws.u.to_uid(src_path)}{ext}', src) 

326 return f'saved {path!r}' 

327 

328 

329def _as_flat_list(ls): 

330 if not isinstance(ls, (list, tuple)): 

331 yield ls 

332 else: 

333 for x in ls: 

334 yield from _as_flat_list(x) 

335 

336 

337def _to_plain(val): 

338 if isinstance(val, (list, tuple)): 

339 return [_to_plain(x) for x in val] 

340 if isinstance(val, gws.Data): 

341 val = vars(val) 

342 if isinstance(val, dict): 

343 return {k: v if k.startswith('_') else _to_plain(v) for k, v in val.items()} 

344 return val