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    ctx.obj['settings_from_arg'] = value == settings_file
161
162    if 'logging_config' in ctx.obj and conf.settings.LOGGING:
163        configure_logging(ctx, None)
164
165
166ConfigAnnotation: TypeAlias = Annotated[
167    typing.Optional[Path],
168    typer.Option(
169        '-c',
170        '--config',
171        callback=load_config,
172        help="Path to [bold blue]config-ninja[/]'s own configuration file.",
173        rich_help_panel='Global',
174        show_default=False,
175    ),
176]
177UserAnnotation: TypeAlias = Annotated[
178    bool,
179    typer.Option(
180        '-u',
181        '--user',
182        '--user-mode',
183        help='User mode installation (does not require [bold orange3]sudo[/])',
184        show_default=False,
185    ),
186]
187WorkdirAnnotation: TypeAlias = Annotated[
188    typing.Optional[Path],
189    typer.Option('-w', '--workdir', help='Run the service from this directory.', show_default=False),
190]
191
192
193def parse_env(ctx: typer.Context, value: typing.Optional[typing.List[str]]) -> typing.List[str]:
194    """Parse the environment variables from the command line."""
195    if ctx.resilient_parsing or not value:
196        return []
197
198    return [v for val in value for v in val.split(',')]
199
200
201EnvNamesAnnotation: TypeAlias = Annotated[
202    typing.Optional[typing.List[str]],
203    typer.Option(
204        '-e',
205        '--env',
206        help='Embed these environment variables into the unit file. Can be used multiple times.',
207        show_default=False,
208        callback=parse_env,
209        metavar='NAME[,NAME...]',
210    ),
211]
212
213
214class UserGroup(typing.NamedTuple):
215    """Run the service using this user (and optionally group)."""
216
217    user: str
218    """The user to run the service as."""
219
220    group: typing.Optional[str] = None
221    """The group to run the service as."""
222
223    @classmethod
224    def parse(cls, value: str) -> 'UserGroup':
225        """Parse the `--run-as user[:group]` argument for the `systemd` service."""
226        return cls(*value.split(':'))
227
228
229RunAsAnnotation: TypeAlias = Annotated[
230    typing.Optional[UserGroup],
231    typer.Option(
232        '--run-as',
233        help='Configure the systemd unit to run the service as this user (and optionally group).',
234        metavar='user[:group]',
235        parser=UserGroup.parse,
236    ),
237]
238
239
240class Variable(typing.NamedTuple):
241    """Set this variable in the shell used to run the `systemd` service."""
242
243    name: str
244    """The name of the variable."""
245
246    value: str
247    """The value of the variable."""
248
249
250def parse_var(value: str) -> Variable:
251    """Parse the `--var VARIABLE=VALUE` arguments for setting variables in the `systemd` service."""
252    try:
253        parsed = Variable(*value.split('='))
254    except TypeError as exc:
255        rich.print(f'[red]ERROR[/]: Invalid argument (expected [yellow]VARIABLE=VALUE[/] pair): [purple]{value}[/]')
256        raise typer.Exit(1) from exc
257
258    return parsed
259
260
261VariableAnnotation: TypeAlias = Annotated[
262    typing.Optional[typing.List[Variable]],
263    typer.Option(
264        '--var',
265        help='Embed the specified [yellow]VARIABLE=VALUE[/] into the unit file. Can be used multiple times.',
266        metavar='VARIABLE=VALUE',
267        show_default=False,
268        parser=parse_var,
269    ),
270]
271
272
273def configure_logging(ctx: typer.Context, verbose: typing.Optional[bool] = None) -> None:
274    """Callback for the `--verbose` option to configure logging verbosity.
275
276    By default, log messages at the `logging.INFO` level:
277
278    >>> configure_logging(ctx)
279    >>> caplog.messages
280    ['logging verbosity set to [green]INFO[/green]']
281
282    <!-- Clear the `caplog` fixture for the `doctest`, but exclude this from the docs
283    >>> caplog.clear()
284
285    -->
286    When `verbose` is `True`, log messages at the `logging.DEBUG` level:
287
288    >>> configure_logging(ctx, True)
289    >>> caplog.messages
290    ['logging verbosity set to [green]DEBUG[/green]']
291    """
292    if ctx.resilient_parsing:  # pragma: no cover  # this is for tab completions
293        return
294
295    ctx.ensure_object(dict)
296
297    # the `--verbose` argument always overrides previous verbosity settings
298    verbose = verbose or ctx.obj.get('verbose')
299    verbosity = logging.DEBUG if verbose else logging.INFO
300
301    logging_config: schema.DictConfigDefault = ctx.obj.get(
302        'logging_config', copy.deepcopy(settings.DEFAULT_LOGGING_CONFIG)
303    )
304
305    conf: typing.Optional[settings.Config] = ctx.obj.get('settings')
306    new_logging_config: schema.DictConfig = (conf.settings.LOGGING or {}) if conf else {}  # type: ignore[assignment,typeddict-item,unused-ignore]
307
308    for key, value in new_logging_config.items():
309        base = logging_config.get(key, {})
310        if isinstance(base, dict):
311            base.update(value)  # type: ignore[call-overload]
312        else:
313            logging_config[key] = value  # type: ignore[literal-required]
314
315    if verbose:
316        logging_config['root']['level'] = verbosity
317        ctx.obj['verbose'] = verbose
318
319    logging.config.dictConfig(logging_config)  # type: ignore[arg-type]
320
321    ctx.obj['logging_config'] = logging_config
322
323    logger.debug(LOG_VERBOSITY_MESSAGE, logging.getLevelName(verbosity), extra={'markup': True})
324
325
326VerbosityAnnotation = Annotated[
327    typing.Optional[bool],
328    typer.Option(
329        '-v',
330        '--verbose',
331        callback=configure_logging,
332        rich_help_panel='Global',
333        help='Log messages at the [black]DEBUG[/] level.',
334        is_eager=True,
335        show_default=False,
336    ),
337]
338
339
340def version_callback(ctx: typer.Context, value: typing.Optional[bool] = None) -> None:
341    """Print the version of the package."""
342    if ctx.resilient_parsing:  # pragma: no cover  # this is for tab completions
343        return
344
345    if value:
346        rich.print(__version__)
347        raise typer.Exit()
348
349
350VersionAnnotation = Annotated[
351    typing.Optional[bool],
352    typer.Option(
353        '-V',
354        '--version',
355        callback=version_callback,
356        rich_help_panel='Global',
357        show_default=False,
358        is_eager=True,
359        help='Print the version and exit.',
360    ),
361]
362
363
364@contextlib.contextmanager
365def handle_key_errors(objects: typing.Dict[str, typing.Any]) -> typing.Iterator[None]:
366    """Handle KeyError exceptions within the managed context."""
367    try:
368        yield
369    except KeyError as exc:  # pragma: no cover
370        rich.print(f'[red]ERROR[/]: Missing key: [green]{exc.args[0]}[/]\n')
371        rich.print(yaml.dump(objects))
372        raise typer.Exit(1) from exc
373
374
375async def poll_all(
376    controllers: typing.List[controller.BackendController], get_or_write: typing.Literal['get', 'write']
377) -> None:
378    """Run the given controllers within an `asyncio` event loop to monitor and apply changes."""
379    await asyncio.gather(*[ctrl.aget(rich.print) if get_or_write == 'get' else ctrl.awrite() for ctrl in controllers])
380
381
382def _check_systemd() -> None:
383    if not systemd.AVAILABLE:
384        rich.print('[red]ERROR[/]: Missing [bold gray93]systemd[/]!')
385        rich.print('Currently, this command only works on linux.')
386        raise typer.Exit(1)
387
388
389# ⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯⎯
390#                                             command definitions
391
392
393@app.command()
394def get(
395    ctx: typer.Context,
396    keys: OptionalKeyAnnotation = None,
397    poll: PollAnnotation = False,
398    get_help: HelpAnnotation = None,
399    config: ConfigAnnotation = None,
400    verbose: VerbosityAnnotation = None,
401    version: VersionAnnotation = None,
402) -> None:
403    """Print the value of the specified configuration object."""
404    conf: settings.Config = ctx.obj['settings']
405
406    controllers = [
407        controller.BackendController.from_settings(conf, key, handle_key_errors)
408        for key in keys or conf.settings.OBJECTS
409    ]
410
411    if poll:
412        logger.debug(
413            'Begin monitoring (read-only): %s',
414            ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers),
415            extra={'markup': True},
416        )
417        asyncio.run(poll_all(controllers, 'get'))
418        return
419
420    for ctrl in controllers:
421        logger.debug('Get [yellow]%s[/yellow]: %s', ctrl.key, ctrl, extra={'markup': True})
422        ctrl.get(rich.print)
423
424
425@app.command()
426def apply(
427    ctx: typer.Context,
428    keys: OptionalKeyAnnotation = None,
429    poll: PollAnnotation = False,
430    get_help: HelpAnnotation = None,
431    config: ConfigAnnotation = None,
432    verbose: VerbosityAnnotation = None,
433    version: VersionAnnotation = None,
434) -> None:
435    """Apply the specified configuration to the system."""
436    conf: settings.Config = ctx.obj['settings']
437    controllers = [
438        controller.BackendController.from_settings(conf, key, handle_key_errors)
439        for key in keys or conf.settings.OBJECTS
440    ]
441
442    if poll:
443        rich.print('Begin monitoring: ' + ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers))
444        asyncio.run(poll_all(controllers, 'write'))
445        return
446
447    for ctrl in controllers:
448        rich.print(f'Apply [yellow]{ctrl.key}[/yellow]: {ctrl}')
449        ctrl.write()
450
451
452@app.command(
453    deprecated=True,
454    short_help='[dim]Apply all configuration objects to the filesystem, and poll for changes.[/] [red](deprecated)[/]',
455    help='Use [bold blue]config-ninja apply --poll[/] instead.',
456)
457def monitor(ctx: typer.Context) -> None:
458    """Apply all configuration objects to the filesystem, and poll for changes."""
459    conf: settings.Config = ctx.obj['settings']
460    controllers = [
461        controller.BackendController.from_settings(conf, key, handle_key_errors) for key in conf.settings.OBJECTS
462    ]
463
464    rich.print('Begin monitoring: ' + ', '.join(f'[yellow]{ctrl.key}[/yellow]' for ctrl in controllers))
465    asyncio.run(poll_all(controllers, 'write'))
466
467
468@app.command()
469def hook(
470    ctx: typer.Context,
471    hook_names: HookAnnotation,
472    get_help: HelpAnnotation = None,
473    config: ConfigAnnotation = None,
474    verbose: VerbosityAnnotation = None,
475    version: VersionAnnotation = None,
476) -> None:
477    """Execute the named hook.
478
479    This command requires the `poe` extra in order to work.
480    """
481    conf: settings.Config = ctx.obj['settings']
482
483    if not conf.engine:
484        fname = ctx.obj.get('settings_file')
485        rich.print(f'[red]ERROR[/]: failed to load hooks from file: [purple]{fname}[/]')
486        raise typer.Exit(1)
487
488    for name in hook_names:
489        conf.engine.get_hook(name)()
490
491
492@self_app.command(name='print')
493def self_print(
494    ctx: typer.Context,
495    get_help: HelpAnnotation = None,
496    config: ConfigAnnotation = None,
497    verbose: VerbosityAnnotation = None,
498    version: VersionAnnotation = None,
499) -> None:
500    """Print [bold blue]config-ninja[/]'s settings."""
501    conf: typing.Optional[settings.Config] = ctx.obj['settings']
502    if not conf:
503        raise typer.Exit(1)
504
505    rich.print(yaml.dump(conf.settings.OBJECTS))
506
507
508@self_app.command()
509def install(
510    ctx: typer.Context,
511    env_names: EnvNamesAnnotation = None,
512    print_only: PrintAnnotation = None,
513    run_as: RunAsAnnotation = None,
514    user_mode: UserAnnotation = False,
515    variables: VariableAnnotation = None,
516    workdir: WorkdirAnnotation = None,
517    get_help: HelpAnnotation = None,
518    config: ConfigAnnotation = None,
519    verbose: VerbosityAnnotation = None,
520    version: VersionAnnotation = None,
521) -> None:
522    """Install [bold blue]config-ninja[/] as a [bold gray93]systemd[/] service.
523
524    Both --env and --var can be passed multiple times.
525
526    Example:
527            config-ninja self install --env FOO,BAR,BAZ --env SPAM --var EGGS=42
528
529    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[/].
530    """
531    environ = {name: os.environ[name] for name in env_names or [] if name in os.environ}
532    environ.update(variables or [])
533
534    settings_file = ctx.obj.get('settings_file')
535    settings_from_arg = ctx.obj.get('settings_from_arg')
536
537    kwargs = {
538        # the command to use when invoking config-ninja from systemd
539        'config_ninja_cmd': sys.argv[0] if sys.argv[0].endswith('config-ninja') else f'{sys.executable} {sys.argv[0]}',
540        # write these environment variables into the systemd service file
541        'environ': environ,
542        # run `config-ninja` from this directory (if specified)
543        'workdir': workdir,
544        'args': f'--config {settings_file}',
545    }
546
547    # override the config file iff it was overridden via the '--config' CLI argument
548    if not settings_from_arg:
549        del kwargs['args']
550
551    if run_as:
552        kwargs['user'] = run_as.user
553        if run_as.group:
554            kwargs['group'] = run_as.group
555
556    svc = systemd.Service('config_ninja', 'systemd.service.j2', user_mode, settings_file if settings_from_arg else None)
557    if print_only:
558        rendered = svc.render(**kwargs)
559        rich.print(Markdown(f'# {svc.path}\n```systemd\n{rendered}\n```'))
560        raise typer.Exit(0)
561
562    _check_systemd()
563
564    rich.print(f'Installing {svc.path}')
565    rich.print(svc.install(**kwargs))
566
567    rich.print('[green]SUCCESS[/] :white_check_mark:')
568
569
570@self_app.command()
571def uninstall(
572    ctx: typer.Context,
573    print_only: PrintAnnotation = None,
574    user: UserAnnotation = False,
575    get_help: HelpAnnotation = None,
576    config: ConfigAnnotation = None,
577    verbose: VerbosityAnnotation = None,
578    version: VersionAnnotation = None,
579) -> None:
580    """Uninstall the [bold blue]config-ninja[/] [bold gray93]systemd[/] service."""
581    settings_file = ctx.obj.get('settings_file') if ctx.obj.get('settings_from_arg') else None
582    svc = systemd.Service('config_ninja', 'systemd.service.j2', user or False, settings_file)
583    if print_only:
584        rich.print(Markdown(f'# {svc.path}\n```systemd\n{svc.read()}\n```'))
585        raise typer.Exit(0)
586
587    _check_systemd()
588
589    rich.print(f'Uninstalling {svc.path}')
590    svc.uninstall()
591    rich.print('[green]SUCCESS[/] :white_check_mark:')
592
593
594@self_app.callback(invoke_without_command=True)
595def self_main(
596    ctx: typer.Context,
597    get_help: HelpAnnotation = None,
598    config: ConfigAnnotation = None,
599    verbose: VerbosityAnnotation = None,
600    version: VersionAnnotation = None,
601) -> None:
602    """Print the help message for the `self` command."""
603    if not ctx.invoked_subcommand:
604        rich.print(ctx.get_help())
605
606
607@app.command()
608def version(
609    ctx: typer.Context,
610    get_help: HelpAnnotation = None,
611    config: ConfigAnnotation = None,
612    verbose: VerbosityAnnotation = None,
613    version: VersionAnnotation = None,
614) -> None:
615    """Print the version and exit."""
616    version_callback(ctx, True)
617
618
619@app.callback(invoke_without_command=True)
620def main(
621    ctx: typer.Context,
622    get_help: HelpAnnotation = None,
623    config: ConfigAnnotation = None,
624    verbose: VerbosityAnnotation = None,
625    version: VersionAnnotation = None,
626) -> None:
627    """Manage operating system configuration files based on data in the cloud."""
628    ctx.ensure_object(dict)
629
630    if not ctx.invoked_subcommand:  # pragma: no cover
631        rich.print(ctx.get_help())
632
633
634logger.debug('successfully imported %s', __name__)