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

1"""Configuration parser: read configuration files and validate them against the specs."""

3from typing import Optional, cast

5import os

6import yaml

8import gws

9import gws.lib.jsonx

10import gws.lib.osx

11import gws.lib.datetimex

12import gws.lib.dynimport

13import gws.lib.vendor.jump

14import gws.lib.vendor.slon

15import gws.spec.runtime

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

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

21 """Read a configuration file and validate it as a given type.

23 Args:

24 path: Path to the configuration file.

25 as_type: Type of the configuration, e.g. ``gws.base.application.core.Config``.

26 ctx: Configuration context. Errors are added to it.

28 Returns:

29 The parsed configuration, or ``None`` if there were errors.

30 """

32 pp = _Parser(ctx)

33 val = pp.read_from_path(path)

34 d = pp.ensure_dict(val, path)

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

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

39 """Validate a configuration dict as a given type.

41 Args:

42 dct: Configuration dict or ``gws.Data`` object.

43 path: Path to the configuration file, for error reporting.

44 as_type: Type of the configuration, e.g. ``gws.ext.config.layer``.

45 ctx: Configuration context. Errors are added to it.

47 Returns:

48 The parsed configuration, or ``None`` if there were errors.

49 """

51 pp = _Parser(ctx)

52 d = pp.ensure_dict(dct, path)

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

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

57 """Read and validate the application configuration file.

59 Sets the server time zone, then parses the application config and all

60 projects (inline ``projects``, ``projectPaths`` and files in ``projectDirs``).

62 Args:

63 path: Path to the application configuration file.

64 ctx: Configuration context. Errors are added to it.

66 Returns:

67 The parsed application configuration, or ``None`` if it could not be parsed.

68 """

70 pp = _Parser(ctx)

71 val = pp.read_from_path(path)

72 d = pp.ensure_dict(val, path)

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

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

77 """Validate an application configuration dict.

79 Works like ``parse_app_from_path``.

81 Args:

82 dct: Application configuration dict or ``gws.Data`` object.

83 path: Path to the configuration file, for error reporting.

84 ctx: Configuration context. Errors are added to it.

86 Returns:

87 The parsed application configuration, or ``None`` if it could not be parsed.

88 """

90 pp = _Parser(ctx)

91 d = pp.ensure_dict(dct, path)

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

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

96 """Read a configuration file into a dict, without validating it.

98 The format is determined by the file extension.

100 Args:

101 path: Path to the configuration file.

102 ctx: Configuration context. Errors are added to it.

103

104 Returns:

105 The configuration dict, or ``None`` if the file could not be read or does not contain a dict.

106 """

107 pp = _Parser(ctx)

108 val = pp.read_from_path(path)

109 d = pp.ensure_dict(val, path)

110 return d

113##

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

117 """Parse the application config and collect all project configs."""

118 dct = gws.u.to_dict(dct)

119 if not isinstance(dct, dict):

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

121 return

122

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

124

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

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

127

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

129 if not app_cfg:

130 return

131

132 projects = []

133 for dcts in inline_projects:

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

135

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

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

138

139 all_project_paths = list(project_paths)

140 for dirname in project_dirs:

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

142

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

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

145

146 app_cfg.set('projectPaths', project_paths)

147 app_cfg.set('projectDirs', project_dirs)

148 app_cfg.set('projects', projects)

149

150 _save_debug(app_cfg, path, '.parsed.json')

151 return app_cfg

152

153

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

155 cfg_list = pp.read_from_path(path)

156 if not cfg_list:

157 return []

158 return _parse_projects(cfg_list, path, pp)

159

160

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

162 """Parse a project config or a (nested) list of them."""

163 ps = []

164

165 for c in _as_flat_list(cfg_list):

166 d = pp.ensure_dict(c, path)

167 if not d:

168 continue

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

170 if prj_cfg:

171 ps.append(prj_cfg)

173 return ps

176##

179class _Parser:

180 """Reads configuration files and validates dicts, recording errors in the context."""

181

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

183 """Initialize the context and enable verbose errors.

184

185 Args:

186 ctx: Configuration context.

187 """

188 self.ctx = ctx

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

190 self.ctx.warnings = ctx.warnings or []

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

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

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

194

195 def ensure_dict(self, val, path):

196 """Convert a value to a plain dict.

197

198 Args:

199 val: Value to convert, a dict or a ``gws.Data`` object.

200 path: Path to the configuration file, for error reporting.

201

202 Returns:

203 A plain dict, or ``None`` if the value is ``None`` or not a dict.

204 """

205 if val is None:

206 return

207 d = _to_plain(val)

208 if not isinstance(d, dict):

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

210 return

211 return d

212

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

214 """Validate a dict against the specs.

215

216 Args:

217 dct: Configuration dict.

218 path: Path to the configuration file, recorded in the context.

219 as_type: Type of the configuration.

220

221 Returns:

222 The parsed configuration, or ``None`` if there were errors.

223 """

224 if not isinstance(dct, dict):

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

226 return

227 if path:

228 _register_path(self.ctx, path)

229 try:

230 cfg = self.ctx.specs.read(

231 dct,

232 as_type,

233 path=path,

234 options=self.ctx.readOptions,

235 )

236 return cast(gws.Config, cfg)

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

238 message, _, cei = exc.args

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

240

241 def read_from_path(self, path: str):

242 """Read a configuration file and convert the result to plain values.

243

244 Args:

245 path: Path to the configuration file.

246

247 Returns:

248 The configuration value, or ``None`` if the file could not be read.

249 """

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

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

252 return

253

254 _register_path(self.ctx, path)

255 r = self.read2(path)

256

257 if r:

258 r = _to_plain(r)

259 _save_debug(r, path, '.src.json')

260 return r

261

262 def read2(self, path: str):

263 """Read a configuration file using the reader for its extension.

264

265 Args:

266 path: Path to the configuration file.

267

268 Returns:

269 The configuration value, or ``None`` on errors.

270 """

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

272 return self.read_py(path)

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

274 return self.read_json(path)

275 if path.endswith(('.yml', '.yaml')):

276 return self.read_yaml(path)

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

278 return self.read_cx(path)

279

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

281

282 def read_py(self, path: str):

283 """Load a Python configuration file and call its ``main`` function with the context.

284

285 Args:

286 path: Path to the configuration file.

287

288 Returns:

289 The value returned by ``main``, or ``None`` on errors.

290 """

291 try:

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

293 if not fn:

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

295 return

296 return fn(self.ctx)

297 except Exception as exc:

298 gws.log.exception()

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

300

301 def read_json(self, path: str):

302 """Read a JSON configuration file.

303

304 Args:

305 path: Path to the configuration file.

306

307 Returns:

308 The decoded value, or ``None`` on errors.

309 """

310 try:

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

312 except Exception as exc:

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

314

315 def read_yaml(self, path: str):

316 """Read a YAML configuration file.

317

318 Args:

319 path: Path to the configuration file.

320

321 Returns:

322 The decoded value, or ``None`` on errors.

323 """

324 try:

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

326 return yaml.safe_load(fp)

327 except Exception as exc:

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

329

330 def read_cx(self, path: str):

331 """Read a ``.cx`` configuration file.

332

333 The file is a ``jump`` template that renders to SLON. Included files are

334 recorded in the context. Template variables are ``true``, ``false``,

335 ``ctx`` and ``gws``.

336

337 Args:

338 path: Path to the configuration file.

339

340 Returns:

341 The decoded value, or ``None`` on errors.

342 """

343 err_cnt = [0]

344

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

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

347 err_cnt[0] += 1

348 return True

349

350 def _loader(cur_path, load_path):

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

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

353 _register_path(self.ctx, load_path)

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

355

356 try:

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

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

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

360 return

361

362 args = {

363 'true': True,

364 'false': False,

365 'ctx': self.ctx,

366 'gws': gws,

367 }

368

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

370 if err_cnt[0] > 0:

371 return

372

373 _save_debug(slon, path, '.src.slon')

374

375 try:

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

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

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

381##

384def _register_path(ctx, path):

385 ctx.paths.add(path)

386

387

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

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

390 cei.message = message

391 cei.update(kwargs)

392 ctx.errors.append(cei)

393

394

395def _register_warning(ctx: gws.ConfigContext, message: str, **kwargs):

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

397 cei.message = message

398 cei.update(kwargs)

399 loc = f' in {cei.path!r}' if cei.path else ''

400 gws.log.warning(f'CONFIGURATION WARNING: {message}{loc}')

401 ctx.warnings.append(cei)

402

403

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

405 """Add a syntax error with the surrounding source lines to the context."""

406 cei = gws.ConfigErrorInfo(

407 path=path,

408 line=line,

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

410 contextLines=[],

411 cause=cause,

412 )

413

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

415 if n < line - context:

416 continue

417 if n > line + context:

418 break

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

420 if n == line:

421 ln = f'>>> {ln}'

422 cei.contextLines.append(ln)

423

424 ctx.errors.append(cei)

425

426

427def _save_debug(src, src_path, ext):

428 """Write an intermediate parsing result to the config directory, for debugging."""

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

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

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

432 return f'saved {path!r}'

433

434

435def _as_flat_list(ls):

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

437 yield ls

438 else:

439 for x in ls:

440 yield from _as_flat_list(x)

441

442

443def _to_plain(val):

444 """Convert ``gws.Data`` objects to dicts recursively, values of keys starting with ``_`` are left as is."""

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

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

447 if isinstance(val, gws.Data):

448 val = vars(val)

449 if isinstance(val, dict):

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

451 return val