Coverage for sygaldry / cli.py: 74%
264 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 18:33 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import asyncio
6import code
7import inspect
8import json
9from pathlib import Path
10from typing import Any, Optional
12import rich_click as click
13import yaml
14from rich.console import Console
15from rich.syntax import Syntax
17from .artificery import Artificery
18from .checker import check as run_check
19from .codegen import generate_check_source
20from .errors import CLIError, SygaldryError
21from .loader import _infer_scalar, _is_url
23_console = Console(stderr=True)
26def _parse_set_option(raw: str) -> tuple[str, Any]:
27 """
28 Parse a ``--set key=value`` string.
30 :param raw: Raw option string.
31 :type raw: str
32 :returns: ``(dotted_path, coerced_value)`` pair.
33 :raises CLIError: If the string has no ``=``.
34 """
35 if "=" not in raw:
36 raise CLIError(f"Invalid --set syntax (expected key=value): '{raw}'")
37 else:
38 key, _, value = raw.partition("=")
39 if key := key.strip():
40 return key, _infer_scalar(value)
41 else:
42 raise CLIError(f"Invalid --set syntax (empty key): '{raw}'")
45def _parse_use_option(raw: str) -> tuple[str, str]:
46 """
47 Parse a ``--use target=source`` string.
49 :param raw: Raw option string.
50 :type raw: str
51 :returns: ``(target_path, source_path)`` pair.
52 :raises CLIError: If the string has no ``=``.
53 """
54 if "=" not in raw:
55 raise CLIError(f"Invalid --use syntax (expected target=source): '{raw}'")
56 target, _, source = raw.partition("=")
57 target = target.strip()
58 source = source.strip()
59 if not target or not source:
60 raise CLIError(f"Invalid --use syntax (empty path): '{raw}'")
61 return target, source
64def _build_artificery(
65 config_paths: tuple[Path, ...],
66 set_overrides: tuple[str, ...],
67 use_overrides: tuple[str, ...],
68 *,
69 cache: Any | None = None,
70 transient: bool = False,
71) -> Artificery:
72 """
73 Build an Artificery from CLI options.
75 :param config_paths: Config file paths.
76 :param set_overrides: ``--set`` option values.
77 :param use_overrides: ``--use`` option values.
78 :param cache: Optional instance cache.
79 :param transient: If True, bypass caching.
80 :returns: Configured Artificery instance.
81 """
82 overrides: dict[str, Any] = dict()
83 for raw in set_overrides:
84 key, value = _parse_set_option(raw)
85 overrides[key] = value
87 uses: dict[str, str] = dict()
88 for raw in use_overrides:
89 target, source = _parse_use_option(raw)
90 uses[target] = source
92 return Artificery(
93 *config_paths,
94 overrides=overrides or None,
95 uses=uses or None,
96 cache=cache,
97 transient=transient,
98 )
101def _extract_call_defaults(
102 config: dict[str, Any], object_key: str
103) -> tuple[str | None, list[Any]]:
104 """
105 Extract ``_call`` defaults from the interpolated config.
107 :param config: Interpolated (pre-resolution) config.
108 :param object_key: Top-level key for the target object.
109 :type config: dict
110 :type object_key: str
111 :returns: ``(method_name_or_None, args_list)`` pair.
112 """
113 obj_config = config.get(object_key)
114 if not isinstance(obj_config, dict):
115 return None, list()
116 call = obj_config.get("_call")
117 if not isinstance(call, dict):
118 return None, list()
119 method = call.get("method")
120 args = call.get("args", [])
121 if not isinstance(args, list): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
122 args = [args]
123 return method, args
126def _parse_method_args(raw_args: tuple[str, ...]) -> list[Any]:
127 """
128 Parse and coerce method arguments from CLI.
130 :param raw_args: Raw argument strings from after ``--``.
131 :type raw_args: tuple[str, ...]
132 :returns: Coerced argument list.
133 """
134 return [_infer_scalar(arg) for arg in raw_args]
137def _invoke_target(obj: Any, method_name: str | None, args: list[Any]) -> Any:
138 """
139 Call a method or the object itself.
141 :param obj: Resolved target object.
142 :param method_name: Method to call, or None for ``__call__``.
143 :param args: Positional arguments.
144 :type obj: object
145 :type method_name: str | None
146 :type args: list
147 :returns: Return value of the call.
148 :raises CLIError: If the object is not callable or method is missing.
149 """
150 if method_name is not None:
151 if not hasattr(obj, method_name):
152 raise CLIError(
153 f"Object {type(obj).__name__!r} has no method '{method_name}'. "
154 f"Available attributes: {sorted(a for a in dir(obj) if not a.startswith('_'))}"
155 )
156 else:
157 target = getattr(obj, method_name)
158 if not callable(target):
159 raise CLIError(
160 f"Attribute '{method_name}' on {type(obj).__name__!r} is not callable."
161 )
162 else:
163 if inspect.iscoroutinefunction(target):
164 return asyncio.run(target(*args))
165 return target(*args)
166 else:
167 if not callable(obj):
168 raise CLIError(
169 f"Object {type(obj).__name__!r} is not callable and no --method was specified."
170 )
172 if inspect.iscoroutinefunction(getattr(obj, "__call__", obj)):
173 return asyncio.run(obj(*args))
174 return obj(*args)
177def _format_config_yaml(config: Any) -> str:
178 """
179 Format a config mapping as YAML.
181 :param config: Config mapping to format.
182 :type config: object
183 :returns: YAML string.
184 """
185 return yaml.dump(config, default_flow_style=False, sort_keys=False)
188def _print_dry_run(
189 config_paths: tuple[Path, ...],
190 object_key: str,
191 method_name: str | None,
192 args: list[Any],
193 config: dict[str, Any],
194) -> None:
195 """
196 Print a dry-run summary.
198 :param config_paths: Config file paths that were loaded.
199 :param object_key: Object key to resolve.
200 :param method_name: Method to call.
201 :param args: Method arguments.
202 :param config: Interpolated config.
203 """
204 _console.print("[bold]Dry run summary[/bold]")
205 _console.print()
206 _console.print(f" Config files: {', '.join(str(p) for p in config_paths)}")
207 _console.print(f" Object: {object_key}")
209 call_desc = method_name or "__call__"
210 _console.print(f" Method: {call_desc}")
212 if args: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true
213 _console.print(f" Args: {args}")
214 else:
215 _console.print(" Args: (none)")
217 obj_config = config.get(object_key, {})
218 if isinstance(obj_config, dict): 218 ↛ exitline 218 didn't return from function '_print_dry_run' because the condition on line 218 was always true
219 _console.print()
220 _console.print(f" [bold]Config for '{object_key}':[/bold]")
221 yaml_str = _format_config_yaml(obj_config)
222 syntax = Syntax(yaml_str, "yaml", theme="monokai", padding=1)
223 _console.print(syntax)
226class _ConfigPathOrURL(click.ParamType):
227 """Click parameter type that accepts a local file path or an HTTP(S) URL."""
229 name = "CONFIG"
231 def convert(self, value, param, ctx):
232 if isinstance(value, Path): 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 return value
234 if _is_url(value): 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true
235 return value
236 path = Path(value)
237 if not path.exists():
238 self.fail(f"Path '{value}' does not exist.", param, ctx)
239 if path.is_dir(): 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 self.fail(f"Path '{value}' is a directory.", param, ctx)
241 return path
244def _config_options(func):
245 """
246 Shared options for config loading, --set, and --use.
247 """
248 func = click.option(
249 "-c",
250 "--config",
251 "config_paths",
252 multiple=True,
253 required=True,
254 type=_ConfigPathOrURL(),
255 envvar="SYGALDRY_CONFIG",
256 help="Config file path or HTTP(S) URL. Repeat for multiple; deep-merged in order.",
257 )(func)
258 func = click.option(
259 "--set",
260 "set_overrides",
261 multiple=True,
262 help="Override a config value: dotted.path=value.",
263 )(func)
264 func = click.option(
265 "--use",
266 "use_overrides",
267 multiple=True,
268 help="Set a config value from another config path: target.path=source.path.",
269 )(func)
270 return func
273@click.group()
274@click.version_option(package_name="sygaldry")
275def cli():
276 """
277 Sygaldry: build and run objects from config files.
278 """
281@cli.command(context_settings={"ignore_unknown_options": True})
282@_config_options
283@click.argument("object_key")
284@click.option(
285 "--method",
286 "method_name",
287 default=None,
288 help="Method to call on the resolved object. Default: call the object itself.",
289)
290@click.option(
291 "--dry-run",
292 is_flag=True,
293 help="Validate and show what would be called, without executing.",
294)
295@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.")
296@click.option("-q", "--quiet", is_flag=True, help="Suppress non-error output.")
297@click.argument("method_args", nargs=-1, type=click.UNPROCESSED)
298def run(
299 config_paths: tuple[Path, ...],
300 set_overrides: tuple[str, ...],
301 use_overrides: tuple[str, ...],
302 object_key: str,
303 method_name: str | None,
304 dry_run: bool,
305 verbose: bool,
306 quiet: bool,
307 method_args: tuple[str, ...],
308) -> None:
309 """
310 Load config, resolve an object, and call it.
311 """
312 try:
313 art = _build_artificery(config_paths, set_overrides, use_overrides)
314 call_method, call_args = _extract_call_defaults(art.config, object_key)
316 final_method = method_name if method_name is not None else call_method
317 final_args = _parse_method_args(method_args) if method_args else call_args
319 if dry_run:
320 _print_dry_run(config_paths, object_key, final_method, final_args, art.config)
321 return
322 else:
323 resolved = art.resolve()
325 if object_key not in resolved:
326 available = sorted(resolved.keys())
327 raise CLIError(
328 f"Object '{object_key}' not found in resolved config. "
329 f"Available keys: {available}"
330 )
332 target = resolved[object_key]
333 result = _invoke_target(target, final_method, final_args)
335 if isinstance(result, int):
336 raise SystemExit(result)
337 if result is not None and not quiet:
338 click.echo(result)
340 except SystemExit:
341 raise
342 except Exception as exc:
343 if verbose: 343 ↛ 344line 343 didn't jump to line 344 because the condition on line 343 was never true
344 _console.print_exception()
345 else:
346 _console.print(f"[bold red]Error:[/bold red] {exc}")
347 raise SystemExit(1) from None
350@cli.command(context_settings={"ignore_unknown_options": True})
351@_config_options
352@click.argument("object_key", required=False, default=None)
353@click.option(
354 "--object",
355 "object_key_opt",
356 default=None,
357 help="Show only this top-level key (also accepted as a positional argument).",
358)
359@click.option(
360 "--method",
361 "method_name",
362 default=None,
363 help="Method to show being called on the resolved object.",
364)
365@click.argument("method_args", nargs=-1, type=click.UNPROCESSED)
366@click.option(
367 "--format",
368 "output_format",
369 type=click.Choice(["yaml", "json"]),
370 default="yaml",
371 help="Output format.",
372)
373@click.option(
374 "--resolved",
375 is_flag=True,
376 help="Show resolved config (objects instantiated).",
377)
378@click.option(
379 "--list-objects",
380 is_flag=True,
381 help="List available top-level config keys.",
382)
383@click.option(
384 "--raw",
385 is_flag=True,
386 help="Show the merged YAML/JSON config instead of generated Python code.",
387)
388@click.option(
389 "-o",
390 "--output",
391 "output_path",
392 type=click.Path(dir_okay=False, path_type=Path),
393 default=None,
394 help="Write output to a file instead of stdout.",
395)
396@click.option(
397 "--line-length",
398 "line_length",
399 type=int,
400 default=99,
401 show_default=True,
402 help="Maximum line length for generated Python code.",
403)
404def show(
405 config_paths: tuple[Path, ...],
406 set_overrides: tuple[str, ...],
407 use_overrides: tuple[str, ...],
408 object_key: str | None,
409 object_key_opt: str | None,
410 method_name: str | None,
411 method_args: tuple[str, ...],
412 output_format: str,
413 resolved: bool,
414 list_objects: bool,
415 raw: bool,
416 output_path: Path | None,
417 line_length: int,
418) -> None:
419 """
420 Display the generated Python code for a config.
422 Optionally pass OBJECT_KEY to also show the call expression for that object.
424 By default, shows the Python code that would be type-checked.
425 Use --raw to see the merged YAML/JSON config instead.
426 """
427 object_key = object_key or object_key_opt
428 try:
429 art = _build_artificery(config_paths, set_overrides, use_overrides)
431 if list_objects:
432 for key in sorted(art.config.keys()):
433 click.echo(key)
434 return
436 if resolved: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true
437 output = art.resolve()
438 if object_key:
439 if object_key not in output:
440 available = sorted(output.keys())
441 raise CLIError(
442 f"Object '{object_key}' not found. Available keys: {available}"
443 )
444 click.echo(repr(output[object_key]))
445 else:
446 for key, value in output.items():
447 click.echo(f"{key}: {repr(value)}")
448 return
450 if raw:
451 display = art.config
452 if object_key:
453 if object_key not in display: 453 ↛ 454line 453 didn't jump to line 454 because the condition on line 453 was never true
454 available = sorted(display.keys())
455 raise CLIError(
456 f"Object '{object_key}' not found. Available keys: {available}"
457 )
458 display = display[object_key]
460 if output_format == "json":
461 text = json.dumps(display, indent=2, default=str)
462 else:
463 text = _format_config_yaml(display)
464 else:
465 call_target = None
466 call_method = None
467 call_args: list[Any] | None = None
469 if object_key: 469 ↛ 470line 469 didn't jump to line 470 because the condition on line 469 was never true
470 call_target = object_key
471 cfg_method, cfg_args = _extract_call_defaults(art.config, object_key)
472 call_method = method_name if method_name is not None else cfg_method
473 call_args = _parse_method_args(method_args) if method_args else cfg_args
475 text, _ = generate_check_source(
476 art.config,
477 line_length=line_length,
478 call_target=call_target,
479 call_method=call_method,
480 call_args=call_args or None,
481 )
483 if output_path: 483 ↛ 484line 483 didn't jump to line 484 because the condition on line 483 was never true
484 output_path.write_text(text)
485 _console.print(f"Wrote output to [bold]{output_path}[/bold]")
486 else:
487 lexer = "yaml" if raw and output_format != "json" else "json" if raw else "python"
488 syntax = Syntax(text, lexer, theme="monokai")
489 Console().print(syntax)
491 except SystemExit:
492 raise
493 except SygaldryError as exc:
494 _console.print(f"[bold red]Error:[/bold red] {exc}")
495 raise SystemExit(1) from None
498@cli.command()
499@_config_options
500def validate(
501 config_paths: tuple[Path, ...],
502 set_overrides: tuple[str, ...],
503 use_overrides: tuple[str, ...],
504) -> None:
505 """
506 Validate config without executing.
507 """
508 try:
509 _build_artificery(config_paths, set_overrides, use_overrides).resolve()
510 _console.print("[bold green]Config is valid.[/bold green]")
512 except SygaldryError as exc:
513 _console.print(f"[bold red]Error:[/bold red] {exc}")
514 raise SystemExit(1) from None
517@cli.command()
518@_config_options
519@click.option(
520 "--type-checker",
521 "type_checker",
522 type=click.Choice(["ty", "basedpyright", "pyright", "mypy"]),
523 default=None,
524 help="Type checker to use (auto-detected if omitted).",
525)
526@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.")
527def check(
528 config_paths: tuple[Path, ...],
529 set_overrides: tuple[str, ...],
530 use_overrides: tuple[str, ...],
531 type_checker: str | None,
532 verbose: bool,
533) -> None:
534 """
535 Static type check a config file.
536 """
538 try:
539 art = _build_artificery(config_paths, set_overrides, use_overrides)
541 errors = run_check(config=art.config, type_checker=type_checker)
543 if not errors:
544 _console.print("[bold green]No type errors found.[/bold green]")
545 return
547 for err in errors:
548 label = err.severity.upper()
549 _console.print(f"[bold red]{label}:[/bold red] {err.config_path}: {err.message}")
551 raise SystemExit(1)
553 except SystemExit:
554 raise
555 except Exception as exc:
556 if verbose:
557 _console.print_exception()
558 else:
559 _console.print(f"[bold red]Error:[/bold red] {exc}")
560 raise SystemExit(1) from None
563@cli.command()
564@_config_options
565@click.option("-v", "--verbose", is_flag=True, help="Show full tracebacks on error.")
566def interactive(
567 config_paths: tuple[Path, ...],
568 set_overrides: tuple[str, ...],
569 use_overrides: tuple[str, ...],
570 verbose: bool,
571) -> None:
572 """
573 Start an interactive Python session with the Artificery loaded.
574 """
575 try:
576 art = _build_artificery(config_paths, set_overrides, use_overrides)
577 # Eagerly prepare the config so errors surface before the REPL starts.
578 _ = art.config
580 except Exception as exc:
581 if verbose:
582 _console.print_exception()
583 else:
584 _console.print(f"[bold red]Error:[/bold red] {exc}")
585 raise SystemExit(1) from None
587 _console.print("[bold green]Sygaldry interactive session[/bold green]")
588 _console.print()
589 _console.print(" Config files: " + ", ".join(str(p) for p in config_paths))
590 _console.print(f" Top-level keys: {sorted(art.config.keys())}")
591 _console.print()
593 namespace: dict[str, Any] = {"artificery": art, "Artificery": Artificery}
595 banner = "\n".join(
596 [
597 "Available variables:",
598 " artificery - Artificery instance (config loaded & merged)",
599 "",
600 "Quick start:",
601 " artificery.config # view the interpolated config",
602 " artificery.resolve() # resolve all objects",
603 ]
604 )
606 try:
607 from IPython import start_ipython
609 start_ipython(argv=[], user_ns=namespace, display_banner=False)
610 except ImportError:
611 code.interact(banner=banner, local=namespace, exitmsg="")
614if __name__ == "__main__": 614 ↛ 615line 614 didn't jump to line 615 because the condition on line 614 was never true
615 cli()