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    'get',
 41    'apply',
 42    'monitor',
 43    'self_print',
 44    'install',
 45    'uninstall',
 46    'version',
 47    'main',
 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    env_names: EnvNamesAnnotation = None,
510    print_only: PrintAnnotation = None,
511    run_as: RunAsAnnotation = None,
512    user_mode: UserAnnotation = False,
513    variables: VariableAnnotation = None,
514    workdir: WorkdirAnnotation = None,
515    get_help: HelpAnnotation = None,
516    config: ConfigAnnotation = None,
517    verbose: VerbosityAnnotation = None,
518    version: VersionAnnotation = None,
519) -> None:
520    """Install [bold blue]config-ninja[/] as a [bold gray93]systemd[/] service.
521
522    Both --env and --var can be passed multiple times.
523
524    Example:
525            config-ninja self install --env FOO,BAR,BAZ --env SPAM --var EGGS=42
526
527    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[/].
528    """
529    environ = {name: os.environ[name] for name in env_names or [] if name in os.environ}
530    environ.update(variables or [])
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    }
540    if run_as:
541        kwargs['user'] = run_as.user
542        if run_as.group:
543            kwargs['group'] = run_as.group
544
545    svc = systemd.Service('config_ninja', 'systemd.service.j2', user_mode)
546    if print_only:
547        rendered = svc.render(**kwargs)
548        rich.print(Markdown(f'# {svc.path}\n```systemd\n{rendered}\n```'))
549        raise typer.Exit(0)
550
551    _check_systemd()
552
553    rich.print(f'Installing {svc.path}')
554    rich.print(svc.install(**kwargs))
555
556    rich.print('[green]SUCCESS[/] :white_check_mark:')
557
558
559@self_app.command()
560def uninstall(
561    print_only: PrintAnnotation = None,
562    user: UserAnnotation = False,
563    get_help: HelpAnnotation = None,
564    config: ConfigAnnotation = None,
565    verbose: VerbosityAnnotation = None,
566    version: VersionAnnotation = None,
567) -> None:
568    """Uninstall the [bold blue]config-ninja[/] [bold gray93]systemd[/] service."""
569    svc = systemd.Service('config_ninja', 'systemd.service.j2', user or False)
570    if print_only:
571        rich.print(Markdown(f'# {svc.path}\n```systemd\n{svc.read()}\n```'))
572        raise typer.Exit(0)
573
574    _check_systemd()
575
576    rich.print(f'Uninstalling {svc.path}')
577    svc.uninstall()
578    rich.print('[green]SUCCESS[/] :white_check_mark:')
579
580
581@self_app.callback(
582    invoke_without_command=True,
583    name='self',
584)
585def self_main(
586    ctx: typer.Context,
587    get_help: HelpAnnotation = None,
588    config: ConfigAnnotation = None,
589    verbose: VerbosityAnnotation = None,
590    version: VersionAnnotation = None,
591) -> None:
592    """Print the help message for the `self` command."""
593    if not ctx.invoked_subcommand:
594        rich.print(ctx.get_help())
595
596
597@app.command()
598def version(
599    ctx: typer.Context,
600    get_help: HelpAnnotation = None,
601    config: ConfigAnnotation = None,
602    verbose: VerbosityAnnotation = None,
603    version: VersionAnnotation = None,
604) -> None:
605    """Print the version and exit."""
606    version_callback(ctx, True)
607
608
609@app.callback(invoke_without_command=True)
610def main(
611    ctx: typer.Context,
612    get_help: HelpAnnotation = None,
613    config: ConfigAnnotation = None,
614    verbose: VerbosityAnnotation = None,
615    version: VersionAnnotation = None,
616) -> None:
617    """Manage operating system configuration files based on data in the cloud."""
618    ctx.ensure_object(dict)
619
620    if not ctx.invoked_subcommand:  # pragma: no cover
621        rich.print(ctx.get_help())
622
623
624logger.debug('successfully imported %s', __name__)