nsconfig/daemon/__init__.py

   1 from io import StringIO
   2 from pathlib import Path
   3 from typing import TextIO
   4 import subprocess
   5 import sys
   6
   7 from nsconfig.core import Nsc, NscZone
   8
   9
  10 class NscDaemon:
  11     nsc: Nsc
  12
  13     def __init__(self) -> None:
  14         pass
  15
  16     def setup(self, nsc: Nsc) -> None:
  17         self.nsc = nsc
  18
  19     def dump_config(self, file: TextIO = sys.stdout) -> None:
  20         pass
  21
  22     def write_config(self) -> None:
  23         pass
  24
  25     def reload_zone(self, z: NscZone) -> None:
  26         pass
  27
  28     def reload_daemon(self) -> None:
  29         pass
  30
  31     def _install_config(self, path: Path, new_contents: str) -> bool:
  32         try:
  33             old_new_contents = path.read_text()
  34         except FileNotFoundError:
  35             old_new_contents = None
  36         if new_contents == old_new_contents:
  37             return False
  38         else:
  39             new_path = Path(str(path) + '.new')
  40             with open(new_path, 'w') as f:
  41                 f.write(new_contents)
  42             new_path.replace(path)
  43             return True
  44
  45     def _write_config(self, config_path: Path) -> bool:
  46         string_stream = StringIO()
  47         self.dump_config(string_stream)
  48         if self._install_config(config_path, string_stream.getvalue()):
  49             print('Wrote new daemon configuration')
  50             return True
  51         else:
  52             print('Daemon configuration not changed')
  53             return False
  54
  55     def _run_command(self, argv, **kwargs) -> None:
  56         res = subprocess.run(argv, **kwargs)
  57         if res.returncode > 0:
  58             print(f'Command failed: {argv}')
  59             sys.exit(1)
  60
  61
  62 class NscDaemonNull(NscDaemon):
  63     pass