Coverage for sygaldry/cli.py: 73%

1from __future__ import annotations

3__author__ = "Rohan B. Dalton"

5import code

6import json

7from pathlib import Path

8from typing import Any, Optional

10import rich_click as click

11import yaml

12from rich.console import Console

13from rich.syntax import Syntax

15from .artificery import Artificery

16from .checker import check as run_check

17from .codegen import generate_check_source

18from .errors import CLIError, SygaldryError

19from .loader import _infer_scalar, _is_url

21_console = Console(stderr=True)

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

25 """

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

28 :param raw: Raw option string.

29 :type raw: str

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

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

32 """

33 if "=" not in raw:

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

35 else:

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

37 if key := key.strip():

38 return key, _infer_scalar(value)

39 else:

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

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

44 """

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

47 :param raw: Raw option string.

48 :type raw: str

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

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

51 """

52 if "=" not in raw:

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

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

55 target = target.strip()

56 source = source.strip()

57 if not target or not source:

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

59 return target, source

62def _build_artificery(

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

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

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

66 *,

67 cache: Any | None = None,

68 transient: bool = False,

69) -> Artificery:

70 """

71 Build an Artificery from CLI options.

73 :param config_paths: Config file paths.

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

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

76 :param cache: Optional instance cache.

77 :param transient: If True, bypass caching.

78 :returns: Configured Artificery instance.

79 """

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

81 for raw in set_overrides:

82 key, value = _parse_set_option(raw)

83 overrides[key] = value

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

86 for raw in use_overrides:

87 target, source = _parse_use_option(raw)

88 uses[target] = source

90 return Artificery(

91 *config_paths,

92 overrides=overrides or None,

93 uses=uses or None,

94 cache=cache,

95 transient=transient,

96 )

99def _extract_call_defaults(

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

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

102 """

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

104

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

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

107 :type config: dict

108 :type object_key: str

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

110 """

111 obj_config = config.get(object_key)

112 if not isinstance(obj_config, dict):

113 return None, list()

114 call = obj_config.get("_call")

115 if not isinstance(call, dict):

116 return None, list()

117 method = call.get("method")

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

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

120 args = [args]

121 return method, args

122

123

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

125 """

126 Parse and coerce method arguments from CLI.

127

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

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

130 :returns: Coerced argument list.

131 """

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

133

134

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

136 """

137 Call a method or the object itself.

138

139 :param obj: Resolved target object.

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

141 :param args: Positional arguments.

142 :type obj: object

143 :type method_name: str | None

144 :type args: list

145 :returns: Return value of the call.

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

147 """

148 if method_name is not None:

149 if not hasattr(obj, method_name):

150 raise CLIError(

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

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

153 )

154 else:

155 target = getattr(obj, method_name)

156 if not callable(target):

157 raise CLIError(

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

159 )

160 else:

161 return target(*args)

162 else:

163 if not callable(obj):

164 raise CLIError(

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

166 )

167

168 return obj(*args)

169

170

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

172 """

173 Format a config mapping as YAML.

174

175 :param config: Config mapping to format.

176 :type config: object

177 :returns: YAML string.

178 """

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

180

181

182def _print_dry_run(

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

184 object_key: str,

185 method_name: str | None,

186 args: list[Any],

187 config: dict[str, Any],

188) -> None:

189 """

190 Print a dry-run summary.

191

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

193 :param object_key: Object key to resolve.

194 :param method_name: Method to call.

195 :param args: Method arguments.

196 :param config: Interpolated config.

197 """

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

199 _console.print()

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

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

202

203 call_desc = method_name or "__call__"

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

205

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

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

208 else:

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

210

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

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

213 _console.print()

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

215 yaml_str = _format_config_yaml(obj_config)

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

217 _console.print(syntax)

218

219

220class _ConfigPathOrURL(click.ParamType):

221 """Click parameter type that accepts a local file path or an HTTP(S) URL."""

222

223 name = "CONFIG"

224

225 def convert(self, value, param, ctx):

226 if isinstance(value, Path): 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true

227 return value

228 if _is_url(value): 228 ↛ 229line 228 didn't jump to line 229 because the condition on line 228 was never true

229 return value

230 path = Path(value)

231 if not path.exists():

232 self.fail(f"Path '{value}' does not exist.", param, ctx)

233 if path.is_dir(): 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 self.fail(f"Path '{value}' is a directory.", param, ctx)

235 return path

236

237

238def _config_options(func):

239 """

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

241 """

242 func = click.option(

243 "-c",

244 "--config",

245 "config_paths",

246 multiple=True,

247 required=True,

248 type=_ConfigPathOrURL(),

249 envvar="SYGALDRY_CONFIG",

250 help="Config file path or HTTP(S) URL. Repeat for multiple; deep-merged in order.",

251 )(func)

252 func = click.option(

253 "--set",

254 "set_overrides",

255 multiple=True,

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

257 )(func)

258 func = click.option(

259 "--use",

260 "use_overrides",

261 multiple=True,

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

263 )(func)

264 return func

265

266

267@click.group()

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

269def cli():

270 """

271 Sygaldry: build and run objects from config files.

272 """

273

274

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

276@_config_options

277@click.argument("object_key")

278@click.option(

279 "--method",

280 "method_name",

281 default=None,

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

283)

284@click.option(

285 "--dry-run",

286 is_flag=True,

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

288)

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

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

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

292def run(

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

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

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

296 object_key: str,

297 method_name: str | None,

298 dry_run: bool,

299 verbose: bool,

300 quiet: bool,

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

302) -> None:

303 """

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

305 """

306 try:

307 art = _build_artificery(config_paths, set_overrides, use_overrides)

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

309

310 final_method = method_name if method_name is not None else call_method

311 final_args = _parse_method_args(method_args) if method_args else call_args

312

313 if dry_run:

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

315 return

316 else:

317 resolved = art.resolve()

318

319 if object_key not in resolved:

320 available = sorted(resolved.keys())

321 raise CLIError(

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

323 f"Available keys: {available}"

324 )

325

326 target = resolved[object_key]

327 result = _invoke_target(target, final_method, final_args)

328

329 if isinstance(result, int):

330 raise SystemExit(result)

331 if result is not None and not quiet:

332 click.echo(result)

333

334 except SystemExit:

335 raise

336 except Exception as exc:

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

338 _console.print_exception()

339 else:

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

341 raise SystemExit(1) from None

342

343

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

345@_config_options

346@click.argument("object_key", required=False, default=None)

347@click.option(

348 "--object",

349 "object_key_opt",

350 default=None,

351 help="Show only this top-level key (also accepted as a positional argument).",

352)

353@click.option(

354 "--method",

355 "method_name",

356 default=None,

357 help="Method to show being called on the resolved object.",

358)

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

360@click.option(

361 "--format",

362 "output_format",

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

364 default="yaml",

365 help="Output format.",

366)

367@click.option(

368 "--resolved",

369 is_flag=True,

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

371)

372@click.option(

373 "--list-objects",

374 is_flag=True,

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

376)

377@click.option(

378 "--raw",

379 is_flag=True,

380 help="Show the merged YAML/JSON config instead of generated Python code.",

381)

382@click.option(

383 "-o",

384 "--output",

385 "output_path",

386 type=click.Path(dir_okay=False, path_type=Path),

387 default=None,

388 help="Write output to a file instead of stdout.",

389)

390@click.option(

391 "--line-length",

392 "line_length",

393 type=int,

394 default=99,

395 show_default=True,

396 help="Maximum line length for generated Python code.",

397)

398def show(

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

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

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

402 object_key: str | None,

403 object_key_opt: str | None,

404 method_name: str | None,

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

406 output_format: str,

407 resolved: bool,

408 list_objects: bool,

409 raw: bool,

410 output_path: Path | None,

411 line_length: int,

412) -> None:

413 """

414 Display the generated Python code for a config.

415

416 Optionally pass OBJECT_KEY to also show the call expression for that object.

417

418 By default, shows the Python code that would be type-checked.

419 Use --raw to see the merged YAML/JSON config instead.

420 """

421 object_key = object_key or object_key_opt

422 try:

423 art = _build_artificery(config_paths, set_overrides, use_overrides)

424

425 if list_objects:

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

427 click.echo(key)

428 return

429

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

431 output = art.resolve()

432 if object_key:

433 if object_key not in output:

434 available = sorted(output.keys())

435 raise CLIError(

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

437 )

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

439 else:

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

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

442 return

443

444 if raw:

445 display = art.config

446 if object_key:

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

448 available = sorted(display.keys())

449 raise CLIError(

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

451 )

452 display = display[object_key]

453

454 if output_format == "json":

455 text = json.dumps(display, indent=2, default=str)

456 else:

457 text = _format_config_yaml(display)

458 else:

459 call_target = None

460 call_method = None

461 call_args: list[Any] | None = None

462

463 if object_key: 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true

464 call_target = object_key

465 cfg_method, cfg_args = _extract_call_defaults(art.config, object_key)

466 call_method = method_name if method_name is not None else cfg_method

467 call_args = _parse_method_args(method_args) if method_args else cfg_args

468

469 text, _ = generate_check_source(

470 art.config,

471 line_length=line_length,

472 call_target=call_target,

473 call_method=call_method,

474 call_args=call_args or None,

475 )

476

477 if output_path: 477 ↛ 478line 477 didn't jump to line 478 because the condition on line 477 was never true

478 output_path.write_text(text)

479 _console.print(f"Wrote output to [bold]{output_path}[/bold]")

480 else:

481 lexer = "yaml" if raw and output_format != "json" else "json" if raw else "python"

482 syntax = Syntax(text, lexer, theme="monokai")

483 Console().print(syntax)

484

485 except SystemExit:

486 raise

487 except SygaldryError as exc:

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

489 raise SystemExit(1) from None

490

491

492@cli.command()

493@_config_options

494def validate(

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

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

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

498) -> None:

499 """

500 Validate config without executing.

501 """

502 try:

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

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

505

506 except SygaldryError as exc:

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

508 raise SystemExit(1) from None

509

510

511@cli.command()

512@_config_options

513@click.option(

514 "--type-checker",

515 "type_checker",

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

517 default=None,

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

519)

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

521def check(

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

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

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

525 type_checker: str | None,

526 verbose: bool,

527) -> None:

528 """

529 Static type check a config file.

530 """

531

532 try:

533 art = _build_artificery(config_paths, set_overrides, use_overrides)

534

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

536

537 if not errors:

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

539 return

540

541 for err in errors:

542 label = err.severity.upper()

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

544

545 raise SystemExit(1)

546

547 except SystemExit:

548 raise

549 except Exception as exc:

550 if verbose:

551 _console.print_exception()

552 else:

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

554 raise SystemExit(1) from None

555

556

557@cli.command()

558@_config_options

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

560def interactive(

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

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

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

564 verbose: bool,

565) -> None:

566 """

567 Start an interactive Python session with the Artificery loaded.

568 """

569 try:

570 art = _build_artificery(config_paths, set_overrides, use_overrides)

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

572 _ = art.config

573

574 except Exception as exc:

575 if verbose:

576 _console.print_exception()

577 else:

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

579 raise SystemExit(1) from None

580

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

582 _console.print()

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

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

585 _console.print()

586

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

588

589 banner = "\n".join(

590 [

591 "Available variables:",

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

593 "",

594 "Quick start:",

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

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

597 ]

598 )

599

600 try:

601 from IPython import start_ipython

602

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

604 except ImportError:

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

606

607

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

609 cli()

Coverage for sygaldry / cli.py: 73%

258 statements