"""A Supervisord client over a unix socket."""
from typing import cast
from pi_portal import config
from pi_portal.modules.python.xmlrpc import patched_client
from pi_portal.modules.system.socket.unix_stream_transport import (
UnixStreamTransport,
)
from pi_portal.modules.system.supervisor_config import (
ProcessList,
ProcessStatus,
)
from typing_extensions import TypedDict
[docs]class SupervisorException(Exception):
"""Exceptions for the SupervisorClient class."""
[docs]class TypeSupervisorProcessInfo(TypedDict):
"""Typed representation of a Supervisor getProcessInfo response."""
statename: ProcessStatus
start: str
[docs]class SupervisorClient:
"""A Supervisor client that connects via the local unix socket.
:param host: Unused, instead the transport uses a unix socket.
:param port: Unused, instead the transport uses a unix socket.
"""
def __init__(self, host: str = 'localhost', port: int = 9001):
self.server = patched_client.Server(
f"http://{host}:{port}",
transport=UnixStreamTransport(config.PATH_SUPERVISOR_SOCKET)
)
[docs] def start(self, process: ProcessList) -> None:
"""Start the specified Supervisor process.
:param process: The process to start.
:raises: :class:`SupervisorException`
"""
try:
self.server.supervisor.startProcess(process.value)
except patched_client.Fault as exc:
raise SupervisorException from exc
[docs] def stop(self, process: ProcessList) -> None:
"""Stop the specified Supervisor process.
:param process: The process to stop.
:raises: :class:`SupervisorException`
"""
try:
self.server.supervisor.stopProcess(process.value)
except patched_client.Fault as exc:
raise SupervisorException from exc
[docs] def status(self, process: ProcessList) -> ProcessStatus:
"""Retrieve the current state of the specified Supervisor process.
:param process: The process to retrieve the status of.
:returns: The status of the queried process.
:raises: :class:`SupervisorException`
"""
try:
return ProcessStatus(
cast(
TypeSupervisorProcessInfo,
self.server.supervisor.getProcessInfo(process.value)
)['statename']
)
except patched_client.Fault as exc:
raise SupervisorException from exc
[docs] def start_time(self, process: ProcessList) -> str:
"""Retrieve the start time of the specified Supervisor process.
:param process: The process to retrieve the uptime of.
:returns: The uptime of the queried process.
:raises: :class:`SupervisorException`
"""
try:
return cast(
TypeSupervisorProcessInfo,
self.server.supervisor.getProcessInfo(process.value),
)['start']
except patched_client.Fault as exc:
raise SupervisorException from exc