1"""Create `config-ninja`_'s CLI with `typer`_.
  2
  3.. include:: cli.md
  4
  5.. note:: `typer`_ does not support `from __future__ import annotations` as of 2023-12-31
  6
  7.. _config-ninja: https://config-ninja.readthedocs.io/home.html
  8.. _typer: https://typer.tiangolo.com/
  9"""
 10
 11import asyncio
 12import contextlib
 13import copy
 14import logging
 15import logging.config
 16import os
 17import sys
 18import typing
 19from pathlib import Path
 20from typing import Annotated, TypeAlias
 21
 22import rich
 23import typer
 24import yaml
 25from rich.markdown import Markdown
 26
 27from config_ninja import __version__, controller, settings, systemd
 28from config_ninja.settings import schema
 29
 30# ruff: noqa: PLR0913
 31# pylint: disable=redefined-outer-name,unused-argument,too-many-arguments
 32
 33__all__ = [
 34    'app',
 35    'apply',
 36    'get',
 37    'install',
 38    'main',
 39    'monitor',
 40    'self_print',
 41    'uninstall',
 42    'version',
 43]
 44
 45LOG_MISSING_SETTINGS_MESSAGE = "Could not find [bold blue]config-ninja[/]'s settings file"
 46LOG_VERBOSITY_MESSAGE = 'logging verbosity set to [green]%s[/green]'
 47
 48logger = logging.getLogger(__name__)
 49
 50app_kwargs: dict[str, typing.Any] = {
 51    'context_settings': {'help_option_names': ['-h', '--help']},
 52    'no_args_is_help': True,
 53    'rich_markup_mode': 'rich',
 54}
 55
 56app = typer.Typer(**app_kwargs)
 57"""The root `typer`_ application.
 58
 59.. _typer: https://typer.tiangolo.com/
 60"""
 61
 62self_app = typer.Typer(**app_kwargs)
 63
 64app.add_typer(self_app, name='self', help='Operate on this installation of [bold blue]config-ninja[/].')
 65
 66ActionType = typing.Callable[[str], typing.Any]
 67
 68
 69def help_callback(ctx: typer.Context, value: bool | None = None) -> None:
 70    """Print the help message for the command."""
 71    if ctx.resilient_parsing:  # pragma: no cover
 72        return
 73
 74    if value:
 75        rich.print(ctx.get_help())
 76        raise typer.Exit()
 77
 78
 79HelpAnnotation: TypeAlias = Annotated[
 80    bool | None,
 81    typer.Option(
 82        '-h',
 83        '--help',
 84        callback=help_callback,
 85        rich_help_panel='Global',
 86        show_default=False,
 87        is_eager=True,
 88        help='Show this message and exit.',
 89    ),
 90]
 91HookAnnotation: TypeAlias = Annotated[
 92    list[str],
 93    typer.Argument(
 94        help='Execute the named hook(s) (multiple values may be provided).',
 95        show_default=False,
 96        metavar='[HOOK...]',
 97    ),
 98]
 99OptionalKeyAnnotation: TypeAlias = Annotated[
100    list[str] | None,
101    typer.Argument(
102        help='Apply the configuration object(s) with matching key(s)'
103        ' (multiple values may be provided). If unspecified, all objects will be applied',
104        show_default=False,
105        metavar='[KEY...]',
106    ),
107]
108PollAnnotation: TypeAlias = Annotated[
109    bool | None,
110    typer.Option(
111        '-p',
112        '--poll',
113        help='Enable polling; print the configuration on changes.',
114        show_default=False,
115    ),
116]
117PrintAnnotation: TypeAlias = Annotated[
118    bool | None,
119    typer.Option(
120        '-p',
121        '--print-only',
122        help='Just print the [bold cyan]config-ninja.service[/] file; do not write.',
123        show_default=False,
124    ),
125]
126
127
128def load_config(ctx: typer.Context, value: Path | None) -> None:
129    """Load the settings file from the given path."""
130    if ctx.resilient_parsing:  # pragma: no cover
131        return
132
133    ctx.ensure_object(dict)
134    if not value and 'settings' in ctx.obj:
135        logger.debug('already loaded settings')
136        return
137
138    try:
139        settings_file = value or settings.resolve_path()
140    except FileNotFoundError as exc:
141        logger.warning(
142            '%s%s',
143            LOG_MISSING_SETTINGS_MESSAGE,
144            (' at any of the following locations:\n  - ' + '\n  - '.join(f'{p}' for p in exc.args[1]))
145            if len(exc.args) > 1
146            else '',
147            extra={'markup': True},
148        )
149        ctx.obj['settings'] = None
150        return
151
152    conf: settings.Config = settings.load(settings_file)
153    ctx.obj['settings'] = conf
154    ctx.obj['settings_file'] = settings_file
155    ctx.obj['settings_from_arg'] = value == settings_file
156
157    if 'logging_config' in ctx.obj and conf.settings.LOGGING:
158        configure_logging(ctx, None)
159
160
161ConfigAnnotation: TypeAlias = Annotated[
162    Path | None,
163    typer.Option(
164        '-c',
165        '--config',
166        callback=load_config,
167        help="Path to [bold blue]config-ninja[/]'s own configuration file.",
168        rich_help_panel='Global',
169        show_default=False,
170    ),
171]
172UserAnnotation: TypeAlias = Annotated[
173    bool,
174    typer.Option(
175        '-u',
176        '--user',
177        '--user-mode',
178        help='User mode installation (does not require [bold orange3]sudo[/])',
179        show_default=False,
180    ),
181]
182WorkdirAnnotation: TypeAlias = Annotated[
183    Path | None,
184    typer.Option('-w', '--workdir', help='Run the service from this directory.', show_default=False),
185]
186
187
188def parse_env(ctx: typer.Context, value: list[str] | None) -> list[str]:
189    """Parse the environment variables from the command line."""
190    if ctx.resilient_parsing or not value:
191        return []
192
193    return [v for val in value for v in val.split(',')]
194
195
196EnvNamesAnnotation: TypeAlias = Annotated[
197    list[str] | None,
198    typer.Option(
199        '-e',
200        '--env',
201        help='Embed these environment variables into the unit file. Can be used multiple times.',
202        show_default=False,
203        callback=parse_env,
204        metavar='NAME[,NAME...]',
205    ),
206]
207
208
209class UserGroup(typing.NamedTuple):
210    """Run the service using this user (and optionally group)."""
211
212    user: str
213    """The user to run the service as."""
214
215    group: str | None = None
216    """The group to run the service as."""
217
218    @classmethod
219    def parse(cls, value: str) -> 'UserGroup':
220        """Parse the `--run-as user[:group]` argument for the `systemd` service."""
221        return cls(*value.split(':'))
222
223
224RunAsAnnotation: TypeAlias = Annotated[
225    UserGroup | None,
226    typer.Option(
227        '--run-as',
228        help='Configure the systemd unit to run the service as this user (and optionally group).',
229        metavar='user[:group]',
230        parser=UserGroup.parse,
231    ),
232]
233
234
235class Variable(typing.NamedTuple):
236    """Set this variable in the shell used to run the `systemd` service."""
237
238    name: str
239    """The name of the variable."""
240
241    value: str
242    """The value of the variable."""
243
244
245def parse_var(value: str) -> Variable:
246    """Parse the `--var VARIABLE=VALUE` arguments for setting variables in the `systemd` service."""
247    try:
248        parsed = Variable(*value.split('='))
249    except TypeError as exc:
250        rich.print(f'[red]ERROR[/]: Invalid argument (expected [yellow]VARIABLE=VALUE[/] pair): [purple]{value}[/]')
251        raise typer.Exit(1) from exc
252
253    return parsed
254
255
256VariableAnnotation: TypeAlias = Annotated[
257    list[Variable] | None,
258    typer.Option(
259        '--var',
260        help='Embed the specified [yellow]VARIABLE=VALUE[/] into the unit file. Can be used multiple times.',
261        metavar='VARIABLE=VALUE',
262        show_default=False,
263        parser=parse_var,
264    ),
265]
266
267
268def configure_logging(ctx: typer.Context, verbose: bool | None = None) -> None:
269    """Callback for the `--verbose` option to configure logging verbosity.
270
271    By default, log messages at the `logging.INFO` level:
272
273    >>> configure_logging(ctx)
274    >>> caplog.messages
275    ['logging verbosity set to [green]INFO[/green]']
276
277    <!-- Clear the `caplog` fixture for the `doctest`, but exclude this from the docs
278    >>> caplog.clear()
279
280    -->
281    When `verbose` is `True`, log messages at the `logging.DEBUG` level:
282
283    >>> configure_logging(ctx, True)
284    >>> caplog.messages
285    ['logging verbosity set to [green]DEBUG[/green]']
286    """
287    if ctx.resilient_parsing:  # pragma: no cover  # this is for tab completions
288        return
289
290    ctx.ensure_object(dict)
291
292    # the `--verbose` argument always overrides previous verbosity settings
293    verbose = verbose or ctx.obj.get('verbose')
294    verbosity = logging.DEBUG if verbose else logging.INFO
295
296    logging_config: schema.DictConfigDefault = ctx.obj.get(
297        'logging_config', copy.deepcopy(settings.DEFAULT_LOGGING_CONFIG)
298    )
299
300    conf: settings.Config | None = ctx.obj.get('settings')
301    new_logging_config: schema.DictConfig = (conf.settings.LOGGING or {}) if conf else {}  # type: ignore[assignment,typeddict-item,unused-ignore]
302
303    for key, value in new_logging_config.items():
304        base = logging_config.get(key, {})
305        if isinstance(base, dict):
306            base.update(value)  # type: ignore[call-overload]
307        else:
308            logging_config[key] = value  # type: ignore[literal-required]
309
310    if verbose:
311        logging_config['root']['level'] = verbosity
312        ctx.obj['verbose'] = verbose
313
314    logging.config.dictConfig(logging_config)  # type: ignore[arg-type]
315
316    ctx.obj['logging_config'] = logging_config
317
318    logger.debug(LOG_VERBOSITY_MESSAGE, logging.getLevelName(verbosity), extra={'markup': True})
319
320
321VerbosityAnnotation = Annotated[
322    bool | None,
323    typer.Option(
324        '-v',
325        '--verbose',
326        callback=configure_logging,
327        rich_help_panel='Global',
328        help='Log messages at the [black]DEBUG[/] level.',
329        is_eager=True,
330        show_default=False,
331    ),
332]
333
334
335def version_callback(ctx: typer.Context, value: bool | None = None) -> None:
336    """Print the version of the package."""
337    if ctx.resilient_parsing:  # pragma: no cover  # this is for tab completions
338        return
339
340    if value:
341        rich.print(__version__)
342        raise typer.Exit()
343
344
345VersionAnnotation = Annotated[
346    bool | None,
347    typer.Option(
348        '-V',
349        '--version',
350        callback=version_callback,
351        rich_help_panel='Global',
352        show_default=False,
353        is_eager=True,
354        help='Print the version and exit.',
355    ),
356]
357
358
359@contextlib.contextmanager
360def handle_key_errors(objects: dict[str, typing.Any]) -> typing.Iterator[None]:
361    """Handle KeyError exceptions within the managed context."""
362    try:
363        yield
364    except KeyError as exc:  # pragma: no cover
365        rich.print(f'[red]ERROR[/]: Missing key: [green]{exc.args[0]}[/]\n')
366        rich.print(yaml.dump(objects))
367        raise typer.Exit(1) from exc
368
369
370async def poll_all(
371    controllers: list[controller.BackendController], get_or_write: typing.Literal['get', 'write']
372) -> None:
373    """Run the given controllers within an `asyncio` event loop to monitor and apply changes."""
374    await asyncio.gather(*[ctrl.aget(rich.print) if get_or_write == 'get' else ctrl.awrite() for ctrl in controllers])
375
376
377def _check_systemd() -> None:
378    if not systemd.AVAILABLE:
379        rich.print('[red]ERROR[/]: Missing [bold gray93]systemd[/]!')
380        rich.print('Currently, this command only works on linux.')
381        raise typer.Exit(1)
382
383
384# ⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯
385#                                             command definitions
386
387
388@app.command()
389def get(
390    ctx: typer.Context,
391    keys: OptionalKeyAnnotation = None,
392    poll: PollAnnotation = False,
393    get_help: HelpAnnotation = None,
394    config: ConfigAnnotation = None,
395    verbose: VerbosityAnnotation = None,
396    version: VersionAnnotation = None,
397) -> None:
398    """Print the value of the specified configuration object."""
399    conf: settings.Config = ctx.obj['settings']
400
401    controllers = [
402        controller.BackendController.from_settings(conf, key, handle_key_errors)
403        for key in keys or conf.settings.OBJECTS
404    ]
405
406    if poll:
407        logger.debug(
408            'Begin monitoring (read-only): %s',
409            ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers),
410            extra={'markup': True},
411        )
412        asyncio.run(poll_all(controllers, 'get'))
413        return
414
415    for ctrl in controllers:
416        logger.debug('Get [yellow]%s[/yellow]: %s', ctrl.key, ctrl, extra={'markup': True})
417        ctrl.get(rich.print)
418
419
420@app.command()
421def apply(
422    ctx: typer.Context,
423    keys: OptionalKeyAnnotation = None,
424    poll: PollAnnotation = False,
425    get_help: HelpAnnotation = None,
426    config: ConfigAnnotation = None,
427    verbose: VerbosityAnnotation = None,
428    version: VersionAnnotation = None,
429) -> None:
430    """Apply the specified configuration to the system."""
431    conf: settings.Config = ctx.obj['settings']
432    controllers = [
433        controller.BackendController.from_settings(conf, key, handle_key_errors)
434        for key in keys or conf.settings.OBJECTS
435    ]
436
437    if poll:
438        rich.print('Begin monitoring: ' + ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers))
439        asyncio.run(poll_all(controllers, 'write'))
440        return
441
442    for ctrl in controllers:
443        rich.print(f'Apply [yellow]{ctrl.key}[/yellow]: {ctrl}')
444        ctrl.write()
445
446
447@app.command(
448    deprecated=True,
449    short_help='[dim]Apply all configuration objects to the filesystem, and poll for changes.[/] [red](deprecated)[/]',
450    help='Use [bold blue]config-ninja apply --poll[/] instead.',
451)
452def monitor(ctx: typer.Context) -> None:
453    """Apply all configuration objects to the filesystem, and poll for changes."""
454    conf: settings.Config = ctx.obj['settings']
455    controllers = [
456        controller.BackendController.from_settings(conf, key, handle_key_errors) for key in conf.settings.OBJECTS
457    ]
458
459    rich.print('Begin monitoring: ' + ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers))
460    asyncio.run(poll_all(controllers, 'write'))
461
462
463@app.command()
464def hook(
465    ctx: typer.Context,
466    hook_names: HookAnnotation,
467    get_help: HelpAnnotation = None,
468    config: ConfigAnnotation = None,
469    verbose: VerbosityAnnotation = None,
470    version: VersionAnnotation = None,
471) -> None:
472    """Execute the named hook.
473
474    This command requires the `poe` extra in order to work.
475    """
476    conf: settings.Config = ctx.obj['settings']
477
478    if not conf.engine:
479        fname = ctx.obj.get('settings_file')
480        rich.print(f'[red]ERROR[/]: failed to load hooks from file: [purple]{fname}[/]')
481        raise typer.Exit(1)
482
483    for name in hook_names:
484        conf.engine.get_hook(name)()
485
486
487@self_app.command(name='print')
488def self_print(
489    ctx: typer.Context,
490    get_help: HelpAnnotation = None,
491    config: ConfigAnnotation = None,
492    verbose: VerbosityAnnotation = None,
493    version: VersionAnnotation = None,
494) -> None:
495    """Print [bold blue]config-ninja[/]'s settings."""
496    conf: settings.Config | None = ctx.obj['settings']
497    if not conf:
498        raise typer.Exit(1)
499
500    rich.print(yaml.dump(conf.settings.OBJECTS))
501
502
503@self_app.command()
504def install(
505    ctx: typer.Context,
506    env_names: EnvNamesAnnotation = None,
507    print_only: PrintAnnotation = None,
508    run_as: RunAsAnnotation = None,
509    user_mode: UserAnnotation = False,
510    variables: VariableAnnotation = None,
511    workdir: WorkdirAnnotation = None,
512    get_help: HelpAnnotation = None,
513    config: ConfigAnnotation = None,
514    verbose: VerbosityAnnotation = None,
515    version: VersionAnnotation = None,
516) -> None:
517    """Install [bold blue]config-ninja[/] as a [bold gray93]systemd[/] service.
518
519    Both --env and --var can be passed multiple times.
520
521    Example:
522            config-ninja self install --env FOO,BAR,BAZ --env SPAM --var EGGS=42
523
524    The environment variables [purple]FOO[/], [purple]BAR[/], [purple]BAZ[/], and [purple]SPAM[/] will be read from the current shell and written to the service file, while [purple]EGGS[/] will be set to [yellow]42[/].
525    """
526    environ = {name: os.environ[name] for name in env_names or [] if name in os.environ}
527    environ.update(variables or [])
528
529    settings_file = ctx.obj.get('settings_file')
530    settings_from_arg = ctx.obj.get('settings_from_arg')
531
532    kwargs = {
533        # the command to use when invoking config-ninja from systemd
534        'config_ninja_cmd': sys.argv[0] if sys.argv[0].endswith('config-ninja') else f'{sys.executable} {sys.argv[0]}',
535        # write these environment variables into the systemd service file
536        'environ': environ,
537        # run `config-ninja` from this directory (if specified)
538        'workdir': workdir,
539        'args': f'--config {settings_file}',
540    }
541
542    # override the config file iff it was overridden via the '--config' CLI argument
543    if not settings_from_arg:
544        del kwargs['args']
545
546    if run_as:
547        kwargs['user'] = run_as.user
548        if run_as.group:
549            kwargs['group'] = run_as.group
550
551    svc = systemd.Service('config_ninja', 'systemd.service.j2', user_mode, settings_file if settings_from_arg else None)
552    if print_only:
553        rendered = svc.render(**kwargs)
554        rich.print(Markdown(f'# {svc.path}\n```systemd\n{rendered}\n```'))
555        raise typer.Exit(0)
556
557    _check_systemd()
558
559    rich.print(f'Installing {svc.path}')
560    rich.print(svc.install(**kwargs))
561
562    rich.print('[green]SUCCESS[/] :white_check_mark:')
563
564
565@self_app.command()
566def uninstall(
567    ctx: typer.Context,
568    print_only: PrintAnnotation = None,
569    user: UserAnnotation = False,
570    get_help: HelpAnnotation = None,
571    config: ConfigAnnotation = None,
572    verbose: VerbosityAnnotation = None,
573    version: VersionAnnotation = None,
574) -> None:
575    """Uninstall the [bold blue]config-ninja[/] [bold gray93]systemd[/] service."""
576    settings_file = ctx.obj.get('settings_file') if ctx.obj.get('settings_from_arg') else None
577    svc = systemd.Service('config_ninja', 'systemd.service.j2', user or False, settings_file)
578    if print_only:
579        rich.print(Markdown(f'# {svc.path}\n```systemd\n{svc.read()}\n```'))
580        raise typer.Exit(0)
581
582    _check_systemd()
583
584    rich.print(f'Uninstalling {svc.path}')
585    svc.uninstall()
586    rich.print('[green]SUCCESS[/] :white_check_mark:')
587
588
589@self_app.callback(invoke_without_command=True)
590def self_main(
591    ctx: typer.Context,
592    get_help: HelpAnnotation = None,
593    config: ConfigAnnotation = None,
594    verbose: VerbosityAnnotation = None,
595    version: VersionAnnotation = None,
596) -> None:
597    """Print the help message for the `self` command."""
598    if not ctx.invoked_subcommand:
599        rich.print(ctx.get_help())
600
601
602@app.command()
603def version(
604    ctx: typer.Context,
605    get_help: HelpAnnotation = None,
606    config: ConfigAnnotation = None,
607    verbose: VerbosityAnnotation = None,
608    version: VersionAnnotation = None,
609) -> None:
610    """Print the version and exit."""
611    version_callback(ctx, True)
612
613
614@app.callback(invoke_without_command=True)
615def main(
616    ctx: typer.Context,
617    get_help: HelpAnnotation = None,
618    config: ConfigAnnotation = None,
619    verbose: VerbosityAnnotation = None,
620    version: VersionAnnotation = None,
621) -> None:
622    """Manage operating system configuration files based on data in the cloud."""
623    ctx.ensure_object(dict)
624
625    if not ctx.invoked_subcommand:  # pragma: no cover
626        rich.print(ctx.get_help())
627
628
629logger.debug('successfully imported %s', __name__)