diff options
Diffstat (limited to 'app.py')
| -rwxr-xr-x | app.py | 170 |
1 files changed, 75 insertions, 95 deletions
@@ -65,10 +65,6 @@ def split_by(xs, sep): yield group -def hostname(): - return socket.gethostname() - - class Response: DEFAULT_STATUS = http.server.HTTPStatus.OK @@ -126,6 +122,32 @@ def run_do(*args, **kwargs): return output.stdout +class Task(abc.ABC): + def complete(self): + self.run() + return Response(Response.body_from_json(self.result())) + + @abc.abstractmethod + def run(self): + pass + + @abc.abstractmethod + def result(self): + pass + + +class TaskList(Task): + def __init__(self, tasks): + self.tasks = tasks + + def run(self): + for task in self.tasks.values(): + task.run() + + def result(self): + return {name: task.result() for name, task in self.tasks.items()} + + pool = ThreadPoolExecutor() @@ -133,13 +155,20 @@ def run(*args, **kwargs): return pool.submit(run_do, *args, **kwargs) -class Command: +class Command(Task): def __init__(self, *args): self.args = args self.env = None def run(self): - return run(*self.args, env=self.env) + self.task = run(*self.args, env=self.env) + return self.task + + def result(self): + return self.task.result() + + def now(self): + return self.run().result() class Systemd(Command): @@ -202,31 +231,19 @@ class Loginctl(Systemd): super().__init__('loginctl', *args, '--full') -class Task(abc.ABC): - def complete(self): - self.run() - return Response(Response.body_from_json(self.result())) - - @abc.abstractmethod +class Hostname(Task): def run(self): pass - @abc.abstractmethod def result(self): - pass + return socket.gethostname() -class TopTask(Task): +class Top(Command): COMMAND = None def __init__(self): - self.cmd = TopTask.get_command() - - def run(self): - self.task = self.cmd.run() - - def result(self): - return self.task.result() + super().__init__(*Top.get_command()) @staticmethod def get_command(): @@ -234,66 +251,43 @@ class TopTask(Task): # from the command line (another option is the rc file, but that's too # complicated). For that, we simply run `top -h` once, and check if # the output contains the flags we want to use. - if TopTask.COMMAND is not None: - return TopTask.COMMAND + if Top.COMMAND is not None: + return Top.COMMAND help_output = run_do('top', '-h') args = ['top', '-b', '-n', '1', '-w', '512'] if 'Ee' in help_output: args += ['-E', 'm', '-e', 'm'] - TopTask.COMMAND = Command(*args) - return TopTask.COMMAND + Top.COMMAND = args + return Top.COMMAND -class RebootTask(Task): +class Reboot(Command): def __init__(self): - self.cmd = Command('systemctl', 'reboot') + super().__init__('systemctl', 'reboot') - def run(self): - self.task = self.cmd.run() - def result(self): - return self.task.result() - - -class PoweroffTask(Task): +class Poweroff(Command): def __init__(self): - self.cmd = Command('systemctl', 'poweroff') + super().__init__('systemctl', 'poweroff') - def run(self): - self.task = self.cmd.run() - - def result(self): - return self.task.result() - -class InstanceStatusTask(Task): +class InstanceStatus(TaskList): def __init__(self, systemctl, journalctl): - self.overview_cmd = systemctl('status') - self.failed_cmd = systemctl('list-units', '--failed') - self.timers_cmd = systemctl('list-timers', '--all') - self.journal_cmd = journalctl('-b', '--lines=20') - - def run(self): - self.overview_task = self.overview_cmd.run() - self.failed_task = self.failed_cmd.run() - self.timers_task = self.timers_cmd.run() - self.journal_task = self.journal_cmd.run() - - def result(self): - return { - 'overview': self.overview_task.result(), - 'failed': self.failed_task.result(), - 'timers': self.timers_task.result(), - 'journal': self.journal_task.result(), + tasks = { + 'overview': systemctl('status'), + 'failed': systemctl('list-units', '--failed'), + 'timers': systemctl('list-timers', '--all'), + 'journal': journalctl('-b', '--lines=20'), } + super().__init__(tasks) -class SystemInstanceStatusTask(InstanceStatusTask): +class SystemStatus(InstanceStatus): def __init__(self): super().__init__(Systemctl.system, Journalctl.system) -class UserInstanceStatusTask(InstanceStatusTask): +class UserStatus(InstanceStatus): def __init__(self, systemctl=Systemctl.user, journalctl=Journalctl.user): super().__init__(systemctl, journalctl) @@ -301,45 +295,31 @@ class UserInstanceStatusTask(InstanceStatusTask): def su(user): systemctl = lambda *args: Systemd.su(user, Systemctl.user(*args)) journalctl = lambda *args: Systemd.su(user, Journalctl.user(*args)) - return UserInstanceStatusTask(systemctl, journalctl) + return UserStatus(systemctl, journalctl) -class UserInstanceStatusTaskList(Task): +class UserStatusList(TaskList): def __init__(self): if running_as_root(): # As root, we can query all the user instances. - self.tasks = {user.name: UserInstanceStatusTask.su(user) for user in systemd_users()} + tasks = {user.name: UserStatus.su(user) for user in systemd_users()} else: # As a regular user, we can only query ourselves. - self.tasks = {} + tasks = {} user = get_current_user() if user_instance_active(user): - self.tasks[user.name] = UserInstanceStatusTask() - - def run(self): - for task in self.tasks.values(): - task.run() - - def result(self): - return {name: task.result() for name, task in self.tasks.items()} + tasks[user.name] = UserStatus() + super().__init__(tasks) -class StatusTask(Task): +class Status(TaskList): def __init__(self): - self.hostname = hostname() - self.system = SystemInstanceStatusTask() - self.user = UserInstanceStatusTaskList() - - def run(self): - self.system.run() - self.user.run() - - def result(self): - return { - 'hostname': self.hostname, - 'system': self.system.result(), - 'user': self.user.result(), + tasks = { + 'hostname': Hostname(), + 'system': SystemStatus(), + 'user': UserStatusList(), } + super().__init__(tasks) User = namedtuple('User', ['uid', 'name']) @@ -373,7 +353,7 @@ def user_instance_active(user): unit_name = f'user@{user.uid}.service' cmd = Systemctl.system('is-active', unit_name, '--quiet') try: - cmd.run().result() + cmd.now() return True except Exception: return False @@ -413,7 +393,7 @@ def human_users(): # that were running a systemd instance. def systemd_users(): def list_users(): - output = Loginctl('list-users', '--no-legend').run().result() + output = Loginctl('list-users', '--no-legend').now() lines = output.splitlines() if not lines: return @@ -432,7 +412,7 @@ def systemd_users(): return None properties = 'UID', 'Name', 'RuntimePath' prop_args = (arg for prop in properties for arg in ('-p', prop)) - output = Loginctl('show-user', *prop_args, '--value', *user_args).run().result() + output = Loginctl('show-user', *prop_args, '--value', *user_args).now() lines = output.splitlines() # Assuming that for muptiple users, the properties will be separated by # an empty string. @@ -462,13 +442,13 @@ class Request(Enum): def process(self): if self is Request.STATUS: - return StatusTask().complete() + return Status().complete() if self is Request.TOP: - return TopTask().complete() + return Top().complete() if self is Request.REBOOT: - return RebootTask().complete() + return Reboot().complete() if self is Request.POWEROFF: - return PoweroffTask().complete() + return Poweroff().complete() raise NotImplementedError(f'unknown request: {self}') |