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    service_name: str
156    """The name of the service to install."""
157
158    valid_chars: str = f'{string.ascii_letters}{string.digits}_-:'
159    """Valid characters for the `systemd` unit file name."""
160
161    max_length: int = 255
162    """Maximum length of the `systemd` unit file name."""
163
164    def __init__(self, provider: str, template: str, user_mode: bool, config_fname: Path | None = None) -> None:
165        """Prepare to render the specified `template` from the `provider` package."""
166        loader = jinja2.PackageLoader(provider)
167        env = jinja2.Environment(autoescape=jinja2.select_autoescape(default=True), loader=loader)
168        self.tmpl = env.get_template(template)
169        self.user_mode = user_mode
170
171        install_path = USER_INSTALL_PATH if user_mode else SYSTEM_INSTALL_PATH
172        if config_fname:
173            base_name = (
174                (
175                    str(config_fname.resolve().with_suffix(''))
176                    .replace('-', '--')
177                    .replace('/', '-')[1 : self.max_length - len('.service')]
178                )
179                + '.service'
180            )
181            self.path = install_path / base_name
182            self.service_name = base_name
183        else:
184            self.path = install_path / 'config-ninja.service'
185            self.service_name = 'config-ninja.service'
186
187        if os.geteuid() == 0:
188            self.sudo = dummy()
189        else:
190            self.sudo = sudo
191
192    def _install_system(self, content: str) -> str:
193        logger.info('writing to %s', self.path)
194        sh.mkdir('-p', str(self.path.parent))
195        sh.tee(str(self.path), _in=content, _out='/dev/null')
196
197        logger.info('enabling and starting %s', self.path.name)
198        sh.systemctl.start(self.path.name)
199        return sh.systemctl.status(self.path.name)
200
201    def _install_user(self, content: str) -> str:
202        logger.info('writing to %s', self.path)
203        self.path.parent.mkdir(parents=True, exist_ok=True)
204        self.path.write_text(content, encoding='utf-8')
205
206        logger.info('enabling and starting %s', self.path.name)
207        sh.systemctl.start('--user', self.path.name)
208        return sh.systemctl.status('--user', self.path.name)
209
210    def _uninstall_system(self) -> None:
211        logger.info('stopping and disabling %s', self.path.name)
212        sh.systemctl.disable('--now', self.path.name)
213
214        logger.info('removing %s', self.path)
215        sh.rm(str(self.path))
216
217    def _uninstall_user(self) -> None:
218        logger.info('stopping and disabling %s', self.path.name)
219        sh.systemctl.disable('--user', '--now', self.path.name)
220
221        logger.info('removing %s', self.path)
222        self.path.unlink()
223
224    def install(self, **kwargs: typing.Any) -> str:
225        """Render the `systemd` service file from `kwargs` and install it."""
226        rendered = self.render(**kwargs)
227        if self.user_mode:
228            return self._install_user(rendered)
229
230        if os.geteuid() == 0:
231            return self._install_system(rendered)
232
233        with sudo:
234            return self._install_system(rendered)
235
236    def read(self) -> str:
237        """Read the `systemd` service file."""
238        return self.path.read_text(encoding='utf-8')
239
240    def render(self, **kwargs: typing.Any) -> str:
241        """Render the `systemd` service file from the given parameters."""
242        if workdir := kwargs.get('workdir'):
243            kwargs['workdir'] = Path(workdir).absolute()
244
245        kwargs.setdefault('service_name', self.service_name)
246
247        kwargs.setdefault('user_mode', self.user_mode)
248
249        return self.tmpl.render(**kwargs)
250
251    def uninstall(self) -> None:
252        """Disable, stop, and delete the service."""
253        if self.user_mode:
254            return self._uninstall_user()
255
256        if os.geteuid() == 0:
257            return self._uninstall_system()
258
259        with sudo:
260            return self._uninstall_system()