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