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