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 string
 44import typing
 45from pathlib import Path
 46from typing import TYPE_CHECKING
 47
 48import jinja2
 49import sdnotify
 50
 51if TYPE_CHECKING:  # pragma: no cover
 52    import sh
 53
 54    AVAILABLE = True
 55else:
 56    try:
 57        import sh
 58    except ImportError:  # pragma: no cover
 59        sh = None
 60        AVAILABLE = False
 61    else:
 62        AVAILABLE = hasattr(sh, 'systemctl')
 63
 64
 65SERVICE_NAME = 'config-ninja.service'
 66SYSTEM_INSTALL_PATH = Path('/etc/systemd/system')
 67"""The file path for system-wide installation."""
 68
 69USER_INSTALL_PATH = Path(os.getenv('XDG_CONFIG_HOME') or Path.home() / '.config') / 'systemd' / 'user'
 70"""The file path for user-local installation."""
 71
 72__all__ = ['SYSTEM_INSTALL_PATH', 'USER_INSTALL_PATH', 'Service', 'notify']
 73logger = logging.getLogger(__name__)
 74
 75
 76@contextlib.contextmanager
 77def dummy() -> typing.Iterator[None]:
 78    """Define a dummy context manager to use instead of `sudo`.
 79
 80    There are a few scenarios where `sudo` is unavailable or unnecessary:
 81    - running on Windows
 82    - running in a container without `sudo` installed
 83    - already running as root
 84    """
 85    yield  # pragma: no cover
 86
 87
 88try:
 89    sudo = sh.contrib.sudo
 90except AttributeError:  # pragma: no cover
 91    sudo = dummy()
 92
 93
 94def notify() -> None:  # pragma: no cover
 95    """Notify `systemd` that the service has finished starting up and is ready."""
 96    sock = sdnotify.SystemdNotifier()
 97    sock.notify('READY=1')  # pyright: ignore[reportUnknownMemberType]
 98
 99
100class Service:
101    """Manipulate the `systemd` service file for `config-ninja`.
102
103    ## User Installation
104
105    To install the service for only the current user, pass `user_mode=True` to the initializer:
106
107    >>> svc = Service('config_ninja', 'systemd.service.j2', user_mode=True)
108    >>> _ = svc.install(
109    ...     config_ninja_cmd='config-ninja', workdir='.', environ={'TESTING': 'true'}
110    ... )
111
112    >>> print(svc.read())
113    [Unit]
114    Description=config synchronization daemon
115    After=network.target
116    <BLANKLINE>
117    [Service]
118    Environment=PYTHONUNBUFFERED=true
119    Environment=TESTING=true
120    ExecStartPre=config-ninja self  print
121    ExecStart=config-ninja apply  --poll
122    Restart=always
123    RestartSec=30s
124    Type=notify
125    WorkingDirectory=...
126    <BLANKLINE>
127    [Install]
128    Alias=config-ninja.service
129
130    >>> svc.uninstall()
131
132    ## System Installation
133
134    For system-wide installation:
135
136    >>> svc = Service('config_ninja', 'systemd.service.j2', user_mode=False)
137    >>> _ = svc.install(
138    ...     config_ninja_cmd='config-ninja', workdir='.', environ={'TESTING': 'true'}
139    ... )
140
141    >>> svc.uninstall()
142    """
143
144    path: Path
145    """The installation location of the `systemd` unit file."""
146
147    sudo: typing.ContextManager[None]
148
149    tmpl: jinja2.Template
150    """Load the template on initialization."""
151
152    user_mode: bool
153    """Whether to install the service for the full system or just the current user."""
154
155    valid_chars: str = f'{string.ascii_letters}{string.digits}_-:'
156    """Valid characters for the `systemd` unit file name."""
157
158    max_length: int = 255
159    """Maximum length of the `systemd` unit file name."""
160
161    def __init__(self, provider: str, template: str, user_mode: bool, config_fname: Path | None = None) -> None:
162        """Prepare to render the specified `template` from the `provider` package."""
163        loader = jinja2.PackageLoader(provider)
164        env = jinja2.Environment(autoescape=jinja2.select_autoescape(default=True), loader=loader)
165        self.tmpl = env.get_template(template)
166        self.user_mode = user_mode
167
168        install_path = USER_INSTALL_PATH if user_mode else SYSTEM_INSTALL_PATH
169        if config_fname:
170            base_name = (
171                (
172                    str(config_fname.resolve().with_suffix(''))
173                    .replace('-', '--')
174                    .replace('/', '-')[1 : self.max_length - len('.service')]
175                )
176                + '.service'
177            )
178            self.path = install_path / base_name
179        else:
180            self.path = install_path / 'config-ninja.service'
181
182        if os.geteuid() == 0:
183            self.sudo = dummy()
184        else:
185            self.sudo = sudo
186
187    def _install_system(self, content: str) -> str:
188        logger.info('writing to %s', self.path)
189        sh.mkdir('-p', str(self.path.parent))
190        sh.tee(str(self.path), _in=content, _out='/dev/null')
191
192        logger.info('enabling and starting %s', self.path.name)
193        sh.systemctl.start(self.path.name)
194        return sh.systemctl.status(self.path.name)
195
196    def _install_user(self, content: str) -> str:
197        logger.info('writing to %s', self.path)
198        self.path.parent.mkdir(parents=True, exist_ok=True)
199        self.path.write_text(content, encoding='utf-8')
200
201        logger.info('enabling and starting %s', self.path.name)
202        sh.systemctl.start('--user', self.path.name)
203        return sh.systemctl.status('--user', self.path.name)
204
205    def _uninstall_system(self) -> None:
206        logger.info('stopping and disabling %s', self.path.name)
207        sh.systemctl.disable('--now', self.path.name)
208
209        logger.info('removing %s', self.path)
210        sh.rm(str(self.path))
211
212    def _uninstall_user(self) -> None:
213        logger.info('stopping and disabling %s', self.path.name)
214        sh.systemctl.disable('--user', '--now', self.path.name)
215
216        logger.info('removing %s', self.path)
217        self.path.unlink()
218
219    def install(self, **kwargs: typing.Any) -> str:
220        """Render the `systemd` service file from `kwargs` and install it."""
221        rendered = self.render(**kwargs)
222        if self.user_mode:
223            return self._install_user(rendered)
224
225        if os.geteuid() == 0:
226            return self._install_system(rendered)
227
228        with sudo:
229            return self._install_system(rendered)
230
231    def read(self) -> str:
232        """Read the `systemd` service file."""
233        return self.path.read_text(encoding='utf-8')
234
235    def render(self, **kwargs: typing.Any) -> str:
236        """Render the `systemd` service file from the given parameters."""
237        if workdir := kwargs.get('workdir'):
238            kwargs['workdir'] = Path(workdir).absolute()
239
240        kwargs.setdefault('user_mode', self.user_mode)
241
242        return self.tmpl.render(**kwargs)
243
244    def uninstall(self) -> None:
245        """Disable, stop, and delete the service."""
246        if self.user_mode:
247            return self._uninstall_user()
248
249        if os.geteuid() == 0:
250            return self._uninstall_system()
251
252        with sudo:
253            return self._uninstall_system()