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