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