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