config_ninja.systemd

  1"""Generate a `systemd` unit file for installation as a service.
  2
  3The following `jinja2` template is used to generate the `systemd` unit file:
  4
  5```jinja
  6.. include:: ./templates/systemd.service.j2
  7```
  8
  9Run the CLI's `install`_ command to install the service:
 10
 11```sh
 12❯ config-ninja self install --env AWS_PROFILE --user
 13Installing /home/ubuntu/.config/systemd/user/config-ninja.service
 14● config-ninja.service - config synchronization daemon
 15     Loaded: loaded (/home/ubuntu/.config/systemd/user/config-ninja.service; disabled; vendor preset: enabled)
 16     Active: active (running) since Sun 2024-01-21 22:37:52 EST; 7ms ago
 17    Process: 20240 ExecStartPre=/usr/local/bin/config-ninja self print (code=exited, status=0/SUCCESS)
 18   Main PID: 20241 (config-ninja)
 19     CGroup: /user.slice/user-1000.slice/user@1000.service/app.slice/config-ninja.service
 20             └─20241 /usr/local/bin/python /usr/local/bin/config-ninja monitor
 21
 22Jan 21 22:37:51 ubuntu config-ninja[20240]:     path: /tmp/config-ninja/settings-subset.toml
 23Jan 21 22:37:51 ubuntu config-ninja[20240]:   source:
 24Jan 21 22:37:51 ubuntu config-ninja[20240]:     backend: local
 25Jan 21 22:37:51 ubuntu config-ninja[20240]:     format: yaml
 26Jan 21 22:37:51 ubuntu config-ninja[20240]:     new:
 27Jan 21 22:37:51 ubuntu config-ninja[20240]:       kwargs:
 28Jan 21 22:37:51 ubuntu config-ninja[20240]:         path: config-ninja-settings.yaml
 29Jan 21 22:37:52 ubuntu config-ninja[20241]: Begin monitoring: ['example-local', 'example-local-template', 'example-appconfig']
 30Jan 21 22:37:52 ubuntu systemd[592]: Started config synchronization daemon.
 31
 32SUCCESS ✅
 33```
 34
 35.. _install: https://bryant-finney.github.io/config-ninja/config_ninja/cli.html#config-ninja-self-install
 36"""  # noqa: RUF002
 37
 38from __future__ import annotations
 39
 40import contextlib
 41import logging
 42import os
 43import typing
 44from pathlib import Path
 45from typing import TYPE_CHECKING
 46
 47import jinja2
 48import sdnotify
 49
 50if TYPE_CHECKING:  # pragma: no cover
 51    import sh
 52else:
 53    try:
 54        import sh
 55    except ImportError:  # pragma: no cover
 56        sh = None
 57
 58# pylint: disable=no-member
 59SERVICE_NAME = 'config-ninja.service'
 60SYSTEM_INSTALL_PATH = Path('/etc/systemd/system')
 61"""The file path for system-wide installation."""
 62
 63USER_INSTALL_PATH = (
 64    Path(os.getenv('XDG_CONFIG_HOME') or Path.home() / '.config') / 'systemd' / 'user'
 65)
 66"""The file path for user-local installation."""
 67
 68__all__ = ['SYSTEM_INSTALL_PATH', 'USER_INSTALL_PATH', 'Service', 'notify']
 69logger = logging.getLogger(__name__)
 70
 71
 72@contextlib.contextmanager
 73def dummy() -> typing.Iterator[None]:
 74    """Define a dummy context manager to use instead of `sudo`.
 75
 76    There are a few scenarios where `sudo` is unavailable or unnecessary:
 77    - running on Windows
 78    - running in a container without `sudo` installed
 79    - already running as root
 80    """
 81    yield  # pragma: no cover
 82
 83
 84try:
 85    sudo = sh.contrib.sudo
 86except AttributeError:  # pragma: no cover
 87    sudo = dummy()
 88
 89
 90def notify() -> None:  # pragma: no cover
 91    """Notify `systemd` that the service has finished starting up and is ready."""
 92    sock = sdnotify.SystemdNotifier()
 93    sock.notify('READY=1')  # pyright: ignore[reportUnknownMemberType]
 94
 95
 96class Service:
 97    """Manipulate the `systemd` service file for `config-ninja`.
 98
 99    ## User Installation
100
101    To install the service for only the current user, pass `user_mode=True` to the initializer:
102
103    >>> svc = Service('config_ninja', 'systemd.service.j2', user_mode=True)
104    >>> _ = svc.install(
105    ...     config_ninja_cmd='config-ninja', workdir='.', environ={'TESTING': 'true'}
106    ... )
107
108    >>> print(svc.read())
109    [Unit]
110    Description=config synchronization daemon
111    After=network.target
112    <BLANKLINE>
113    [Service]
114    Environment=PYTHONUNBUFFERED=true
115    Environment=TESTING=true
116    ExecStartPre=config-ninja self print
117    ExecStart=config-ninja monitor
118    Restart=always
119    RestartSec=30s
120    Type=notify
121    WorkingDirectory=...
122    <BLANKLINE>
123    [Install]
124    Alias=config-ninja.service
125
126    >>> svc.uninstall()
127
128    ## System Installation
129
130    For system-wide installation:
131
132    >>> svc = Service('config_ninja', 'systemd.service.j2', user_mode=False)
133    >>> _ = svc.install(
134    ...     config_ninja_cmd='config-ninja', workdir='.', environ={'TESTING': 'true'}
135    ... )
136
137    >>> svc.uninstall()
138    """
139
140    path: Path
141    """The installation location of the `systemd` unit file."""
142
143    sudo: typing.ContextManager[None]
144
145    tmpl: jinja2.Template
146    """Load the template on initialization."""
147
148    user_mode: bool
149    """Whether to install the service for the full system or just the current user."""
150
151    def __init__(self, provider: str, template: str, user_mode: bool) -> None:
152        """Prepare to render the specified `template` from the `provider` package."""
153        loader = jinja2.PackageLoader(provider)
154        env = jinja2.Environment(autoescape=jinja2.select_autoescape(default=True), loader=loader)
155        self.tmpl = env.get_template(template)
156        self.user_mode = user_mode
157        self.path = (USER_INSTALL_PATH if user_mode else SYSTEM_INSTALL_PATH) / SERVICE_NAME
158
159        if os.geteuid() == 0:
160            self.sudo = dummy()
161        else:
162            self.sudo = sudo
163
164    def _install_system(self, content: str) -> str:
165        logger.info('writing to %s', self.path)
166        sh.mkdir('-p', str(self.path.parent))
167        sh.tee(str(self.path), _in=content, _out='/dev/null')
168
169        logger.info('enabling and starting %s', self.path.name)
170        sh.systemctl.start(self.path.name)
171        return sh.systemctl.status(self.path.name)
172
173    def _install_user(self, content: str) -> str:
174        logger.info('writing to %s', self.path)
175        self.path.parent.mkdir(parents=True, exist_ok=True)
176        self.path.write_text(content, encoding='utf-8')
177
178        logger.info('enabling and starting %s', self.path.name)
179        sh.systemctl.start('--user', self.path.name)
180        return sh.systemctl.status('--user', self.path.name)
181
182    def _uninstall_system(self) -> None:
183        logger.info('stopping and disabling %s', self.path.name)
184        sh.systemctl.disable('--now', self.path.name)
185
186        logger.info('removing %s', self.path)
187        sh.rm(str(self.path))
188
189    def _uninstall_user(self) -> None:
190        logger.info('stopping and disabling %s', self.path.name)
191        sh.systemctl.disable('--user', '--now', self.path.name)
192
193        logger.info('removing %s', self.path)
194        self.path.unlink()
195
196    def install(self, **kwargs: typing.Any) -> str:
197        """Render the `systemd` service file from `kwargs` and install it."""
198        rendered = self.render(**kwargs)
199        if self.user_mode:
200            return self._install_user(rendered)
201
202        if os.geteuid() == 0:
203            return self._install_system(rendered)
204
205        with sudo:
206            return self._install_system(rendered)
207
208    def read(self) -> str:
209        """Read the `systemd` service file."""
210        return self.path.read_text(encoding='utf-8')
211
212    def render(self, **kwargs: typing.Any) -> str:
213        """Render the `systemd` service file from the given parameters."""
214        if workdir := kwargs.get('workdir'):
215            kwargs['workdir'] = Path(workdir).absolute()
216
217        kwargs.setdefault('user_mode', self.user_mode)
218
219        return self.tmpl.render(**kwargs)
220
221    def uninstall(self) -> None:
222        """Disable, stop, and delete the service."""
223        if self.user_mode:
224            return self._uninstall_user()
225
226        if os.geteuid() == 0:
227            return self._uninstall_system()
228
229        with sudo:
230            return self._uninstall_system()