Coverage for sygaldry / cli.py: 73%

223 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-04-01 00:11 -0400

1from __future__ import annotations 

2 

3__author__ = "Rohan B. Dalton" 

4 

5import code 

6import json 

7from pathlib import Path 

8from typing import Any, Optional 

9 

10import rich_click as click 

11import yaml 

12from rich.console import Console 

13from rich.syntax import Syntax 

14 

15from .artificery import Artificery 

16from .checker import check as run_check 

17from .errors import CLIError, SygaldryError 

18from .loader import _infer_scalar 

19 

20_console = Console(stderr=True) 

21 

22 

23def _parse_set_option(raw: str) -> tuple[str, Any]: 

24 """ 

25 Parse a ``--set key=value`` string. 

26 

27 :param raw: Raw option string. 

28 :type raw: str 

29 :returns: ``(dotted_path, coerced_value)`` pair. 

30 :raises CLIError: If the string has no ``=``. 

31 """ 

32 if "=" not in raw: 

33 raise CLIError(f"Invalid --set syntax (expected key=value): '{raw}'") 

34 else: 

35 key, _, value = raw.partition("=") 

36 if key := key.strip(): 

37 return key, _infer_scalar(value) 

38 else: 

39 raise CLIError(f"Invalid --set syntax (empty key): '{raw}'") 

40 

41 

42def _parse_use_option(raw: str) -> tuple[str, str]: 

43 """ 

44 Parse a ``--use target=source`` string. 

45 

46 :param raw: Raw option string. 

47 :type raw: str 

48 :returns: ``(target_path, source_path)`` pair. 

49 :raises CLIError: If the string has no ``=``. 

50 """ 

51 if "=" not in raw: 

52 raise CLIError(f"Invalid --use syntax (expected target=source): '{raw}'") 

53 target, _, source = raw.partition("=") 

54 target = target.strip() 

55 source = source.strip() 

56 if not target or not source: 

57 raise CLIError(f"Invalid --use syntax (empty path): '{raw}'") 

58 return target, source 

59 

60 

61def _build_artificery( 

62 config_paths: tuple[Path, ...], 

63 set_overrides: tuple[str, ...], 

64 use_overrides: tuple[str, ...], 

65 *, 

66 cache: Any | None = None, 

67 transient: bool = False, 

68) -> Artificery: 

69 """ 

70 Build an Artificery from CLI options. 

71 

72 :param config_paths: Config file paths. 

73 :param set_overrides: ``--set`` option values. 

74 :param use_overrides: ``--use`` option values. 

75 :param cache: Optional instance cache. 

76 :param transient: If True, bypass caching. 

77 :returns: Configured Artificery instance. 

78 """ 

79 overrides: dict[str, Any] = dict() 

80 for raw in set_overrides: 

81 key, value = _parse_set_option(raw) 

82 overrides[key] = value 

83 

84 uses: dict[str, str] = dict() 

85 for raw in use_overrides: 

86 target, source = _parse_use_option(raw) 

87 uses[target] = source 

88 

89 return Artificery( 

90 *config_paths, 

91 overrides=overrides or None, 

92 uses=uses or None, 

93 cache=cache, 

94 transient=transient, 

95 ) 

96 

97 

98def _extract_call_defaults( 

99 config: dict[str, Any], object_key: str 

100) -> tuple[str | None, list[Any]]: 

101 """ 

102 Extract ``_call`` defaults from the interpolated config. 

103 

104 :param config: Interpolated (pre-resolution) config. 

105 :param object_key: Top-level key for the target object. 

106 :type config: dict 

107 :type object_key: str 

108 :returns: ``(method_name_or_None, args_list)`` pair. 

109 """ 

110 obj_config = config.get(object_key) 

111 if not isinstance(obj_config, dict): 

112 return None, list() 

113 call = obj_config.get("_call") 

114 if not isinstance(call, dict): 

115 return None, list() 

116 method = call.get("method") 

117 args = call.get("args", []) 

118 if not isinstance(args, list): 118 ↛ 119line 118 didn't jump to line 119 because the condition on line 118 was never true

119 args = [args] 

120 return method, args 

121 

122 

123def _parse_method_args(raw_args: tuple[str, ...]) -> list[Any]: 

124 """ 

125 Parse and coerce method arguments from CLI. 

126 

127 :param raw_args: Raw argument strings from after ``--``. 

128 :type raw_args: tuple[str, ...] 

129 :returns: Coerced argument list. 

130 """ 

131 return [_infer_scalar(arg) for arg in raw_args] 

132 

133 

134def _invoke_target(obj: Any, method_name: str | None, args: list[Any]) -> Any: 

135 """ 

136 Call a method or the object itself. 

137 

138 :param obj: Resolved target object. 

139 :param method_name: Method to call, or None for ``__call__``. 

140 :param args: Positional arguments. 

141 :type obj: object 

142 :type method_name: str | None 

143 :type args: list 

144 :returns: Return value of the call. 

145 :raises CLIError: If the object is not callable or method is missing. 

146 """ 

147 if method_name is not None: 

148 if not hasattr(obj, method_name): 

149 raise CLIError( 

150 f"Object {type(obj).__name__!r} has no method '{method_name}'. " 

151 f"Available attributes: {sorted(a for a in dir(obj) if not a.startswith('_'))}" 

152 ) 

153 else: 

154 target = getattr(obj, method_name) 

155 if not callable(target): 

156 raise CLIError( 

157 f"Attribute '{method_name}' on {type(obj).__name__!r} is not callable." 

158 ) 

159 else: 

160 return target(*args) 

161 else: 

162 if not callable(obj): 

163 raise CLIError( 

164 f"Object {type(obj).__name__!r} is not callable and no --method was specified." 

165 ) 

166 

167 return obj(*args) 

168 

169 

170def _format_config_yaml(config: Any) -> str: 

171 """ 

172 Format a config mapping as YAML. 

173 

174 :param config: Config mapping to format. 

175 :type config: object 

176 :returns: YAML string. 

177 """ 

178 return yaml.dump(config, default_flow_style=False, sort_keys=False) 

179 

180 

181def _print_dry_run( 

182 config_paths: tuple[Path, ...], 

183 object_key: str, 

184 method_name: str | None, 

185 args: list[Any], 

186 config: dict[str, Any], 

187) -> None: 

188 """Print a dry-run summary. 

189 

190 :param config_paths: Config file paths that were loaded. 

191 :param object_key: Object key to resolve. 

192 :param method_name: Method to call. 

193 :param args: Method arguments. 

194 :param config: Interpolated config. 

195 """ 

196 _console.print("[bold]Dry run summary[/bold]") 

197 _console.print() 

198 _console.print(f" Config files: {', '.join(str(p) for p in config_paths)}") 

199 _console.print(f" Object: {object_key}") 

200 

201 call_desc = method_name or "__call__" 

202 _console.print(f" Method: {call_desc}") 

203 

204 if args: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true

205 _console.print(f" Args: {args}") 

206 else: 

207 _console.print(" Args: (none)") 

208 

209 obj_config = config.get(object_key, {}) 

210 if isinstance(obj_config, dict): 210 ↛ exitline 210 didn't return from function '_print_dry_run' because the condition on line 210 was always true

211 _console.print() 

212 _console.print(f" [bold]Config for '{object_key}':[/bold]") 

213 yaml_str = _format_config_yaml(obj_config) 

214 syntax = Syntax(yaml_str, "yaml", theme="monokai", padding=1) 

215 _console.print(syntax) 

216 

217 

218def _config_options(func): 

219 """ 

220 Shared options for config loading, --set, and --use. 

221 """ 

222 func = click.option( 

223 "-c", 

224 "--config", 

225 "config_paths", 

226 multiple=True, 

227 required=True, 

228 type=click.Path(exists=True, dir_okay=False, path_type=Path), 

229 envvar="SYGALDRY_CONFIG", 

230 help="Config file path. Repeat for multiple files; deep-merged in order.", 

231 )(func) 

232 func = click.option( 

233 "--set", 

234 "set_overrides", 

235 multiple=True, 

236 help="Override a config value: dotted.path=value.", 

237 )(func) 

238 func = click.option( 

239 "--use", 

240 "use_overrides", 

241 multiple=True, 

242 help="Set a config value from another config path: target.path=source.path.", 

243 )(func) 

244 return func 

245 

246 

247@click.group() 

248@click.version_option(package_name="sygaldry") 

249def cli(): 

250 """ 

251 Sygaldry: build and run objects from config files. 

252 """ 

253 

254 

255@cli.command(context_settings={"ignore_unknown_options": True}) 

256@_config_options 

257@click.argument("object_key") 

258@click.option( 

259 "--method", 

260 "method_name", 

261 default=None, 

262 help="Method to call on the resolved object. Default: call the object itself.", 

263) 

264@click.option( 

265 "--dry-run", 

266 is_flag=True, 

267 help="Validate and show what would be called, without executing.", 

268) 

269@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.") 

270@click.option("-q", "--quiet", is_flag=True, help="Suppress non-error output.") 

271@click.argument("method_args", nargs=-1, type=click.UNPROCESSED) 

272def run( 

273 config_paths: tuple[Path, ...], 

274 set_overrides: tuple[str, ...], 

275 use_overrides: tuple[str, ...], 

276 object_key: str, 

277 method_name: str | None, 

278 dry_run: bool, 

279 verbose: bool, 

280 quiet: bool, 

281 method_args: tuple[str, ...], 

282) -> None: 

283 """ 

284 Load config, resolve an object, and call it. 

285 """ 

286 try: 

287 art = _build_artificery(config_paths, set_overrides, use_overrides) 

288 call_method, call_args = _extract_call_defaults(art.config, object_key) 

289 

290 final_method = method_name if method_name is not None else call_method 

291 final_args = _parse_method_args(method_args) if method_args else call_args 

292 

293 if dry_run: 

294 _print_dry_run(config_paths, object_key, final_method, final_args, art.config) 

295 return 

296 else: 

297 resolved = art.resolve() 

298 

299 if object_key not in resolved: 

300 available = sorted(resolved.keys()) 

301 raise CLIError( 

302 f"Object '{object_key}' not found in resolved config. " 

303 f"Available keys: {available}" 

304 ) 

305 

306 target = resolved[object_key] 

307 result = _invoke_target(target, final_method, final_args) 

308 

309 if isinstance(result, int): 

310 raise SystemExit(result) 

311 if result is not None and not quiet: 

312 click.echo(result) 

313 

314 except SystemExit: 

315 raise 

316 except Exception as exc: 

317 if verbose: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true

318 _console.print_exception() 

319 else: 

320 _console.print(f"[bold red]Error:[/bold red] {exc}") 

321 raise SystemExit(1) from None 

322 

323 

324@cli.command() 

325@_config_options 

326@click.option( 

327 "--object", 

328 "object_key", 

329 default=None, 

330 help="Show only this top-level key.", 

331) 

332@click.option( 

333 "--format", 

334 "output_format", 

335 type=click.Choice(["yaml", "json"]), 

336 default="yaml", 

337 help="Output format.", 

338) 

339@click.option( 

340 "--resolved", 

341 is_flag=True, 

342 help="Show resolved config (objects instantiated).", 

343) 

344@click.option( 

345 "--list-objects", 

346 is_flag=True, 

347 help="List available top-level config keys.", 

348) 

349def show( 

350 config_paths: tuple[Path, ...], 

351 set_overrides: tuple[str, ...], 

352 use_overrides: tuple[str, ...], 

353 object_key: str | None, 

354 output_format: str, 

355 resolved: bool, 

356 list_objects: bool, 

357) -> None: 

358 """Display the merged config for debugging.""" 

359 try: 

360 art = _build_artificery(config_paths, set_overrides, use_overrides) 

361 

362 if list_objects: 

363 for key in sorted(art.config.keys()): 

364 click.echo(key) 

365 return 

366 

367 if resolved: 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true

368 output = art.resolve() 

369 if object_key: 

370 if object_key not in output: 

371 available = sorted(output.keys()) 

372 raise CLIError( 

373 f"Object '{object_key}' not found. Available keys: {available}" 

374 ) 

375 click.echo(repr(output[object_key])) 

376 else: 

377 for key, value in output.items(): 

378 click.echo(f"{key}: {repr(value)}") 

379 return 

380 

381 display = art.config 

382 if object_key: 

383 if object_key not in display: 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true

384 available = sorted(display.keys()) 

385 raise CLIError(f"Object '{object_key}' not found. Available keys: {available}") 

386 display = display[object_key] 

387 

388 if output_format == "json": 

389 click.echo(json.dumps(display, indent=2, default=str)) 

390 else: 

391 yaml_str = _format_config_yaml(display) 

392 syntax = Syntax(yaml_str, "yaml", theme="monokai") 

393 Console().print(syntax) 

394 

395 except SystemExit: 

396 raise 

397 except SygaldryError as exc: 

398 _console.print(f"[bold red]Error:[/bold red] {exc}") 

399 raise SystemExit(1) from None 

400 

401 

402@cli.command() 

403@_config_options 

404def validate( 

405 config_paths: tuple[Path, ...], 

406 set_overrides: tuple[str, ...], 

407 use_overrides: tuple[str, ...], 

408) -> None: 

409 """Validate config without executing.""" 

410 try: 

411 _build_artificery(config_paths, set_overrides, use_overrides).resolve() 

412 _console.print("[bold green]Config is valid.[/bold green]") 

413 

414 except SygaldryError as exc: 

415 _console.print(f"[bold red]Error:[/bold red] {exc}") 

416 raise SystemExit(1) from None 

417 

418 

419@cli.command() 

420@_config_options 

421@click.option( 

422 "--type-checker", 

423 "type_checker", 

424 type=click.Choice(["ty", "basedpyright", "pyright", "mypy"]), 

425 default=None, 

426 help="Type checker to use (auto-detected if omitted).", 

427) 

428@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.") 

429def check( 

430 config_paths: tuple[Path, ...], 

431 set_overrides: tuple[str, ...], 

432 use_overrides: tuple[str, ...], 

433 type_checker: str | None, 

434 verbose: bool, 

435) -> None: 

436 """ 

437 Static type check a config file. 

438 """ 

439 

440 try: 

441 art = _build_artificery(config_paths, set_overrides, use_overrides) 

442 

443 errors = run_check(config=art.config, type_checker=type_checker) 

444 

445 if not errors: 

446 _console.print("[bold green]No type errors found.[/bold green]") 

447 return 

448 

449 for err in errors: 

450 label = err.severity.upper() 

451 _console.print(f"[bold red]{label}:[/bold red] {err.config_path}: {err.message}") 

452 

453 raise SystemExit(1) 

454 

455 except SystemExit: 

456 raise 

457 except Exception as exc: 

458 if verbose: 

459 _console.print_exception() 

460 else: 

461 _console.print(f"[bold red]Error:[/bold red] {exc}") 

462 raise SystemExit(1) from None 

463 

464 

465@cli.command() 

466@_config_options 

467@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.") 

468def interactive( 

469 config_paths: tuple[Path, ...], 

470 set_overrides: tuple[str, ...], 

471 use_overrides: tuple[str, ...], 

472 verbose: bool, 

473) -> None: 

474 """ 

475 Start an interactive Python session with the Artificery loaded. 

476 """ 

477 try: 

478 art = _build_artificery(config_paths, set_overrides, use_overrides) 

479 # Eagerly prepare the config so errors surface before the REPL starts. 

480 _ = art.config 

481 

482 except Exception as exc: 

483 if verbose: 

484 _console.print_exception() 

485 else: 

486 _console.print(f"[bold red]Error:[/bold red] {exc}") 

487 raise SystemExit(1) from None 

488 

489 _console.print("[bold green]Sygaldry interactive session[/bold green]") 

490 _console.print() 

491 _console.print(" Config files: " + ", ".join(str(p) for p in config_paths)) 

492 _console.print(f" Top-level keys: {sorted(art.config.keys())}") 

493 _console.print() 

494 

495 namespace: dict[str, Any] = {"artificery": art, "Artificery": Artificery} 

496 

497 banner = "\n".join( 

498 [ 

499 "Available variables:", 

500 " artificery - Artificery instance (config loaded & merged)", 

501 "", 

502 "Quick start:", 

503 " artificery.config # view the interpolated config", 

504 " artificery.resolve() # resolve all objects", 

505 ] 

506 ) 

507 

508 try: 

509 from IPython import start_ipython 

510 

511 start_ipython(argv=[], user_ns=namespace, display_banner=False) 

512 except ImportError: 

513 code.interact(banner=banner, local=namespace, exitmsg="") 

514 

515 

516if __name__ == "__main__": 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true

517 cli()