Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
2.4 kB · 91 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192from collections.abc import Coroutinefrom dataclasses import dataclass
import asyncioimport enumimport typing
from dbus_fast.aio import MessageBus, ProxyInterface
SYSTEMD_SERVICE = 'org.freedesktop.systemd1'SYSTEMD_ROOT = '/org/freedesktop/systemd1'
class StartMode(enum.Enum): REPLACE = enum.auto() FAIL = enum.auto() ISOLATE = enum.auto() IGNORE_DEPENDENCIES = enum.auto() IGNORE_REQUIREMENTS = enum.auto()
class JobResult(enum.Enum): DONE = enum.auto() CANCELED = enum.auto() TIMEOUT = enum.auto() FAILED = enum.auto() DEPENDENCY = enum.auto() SKIPPED = enum.auto()
@dataclassclass Job: path: str
@dataclass(kw_only=True)class Manager: INTERFACE: typing.ClassVar[str] = 'org.freedesktop.systemd1.Manager'
proxy: ProxyInterface
async def restart_unit(self, unit: str, mode: StartMode) -> Job: job = typing.cast( 'str', await typing.cast( 'typing.Any', self.proxy, ).call_restart_unit( # pyright: ignore[reportAny] unit, mode.name.lower().replace('_', '-'), ), ) return Job(job)
async def wait_for_job_removal( self, cb: Coroutine[typing.Any, typing.Any, Job], ) -> JobResult: queue = asyncio.Queue[tuple[str, str]]()
def on_job_removed(_id: int, job: str, _unit: str, result: str) -> None: queue.put_nowait((job, result))
typing.cast('typing.Any', self.proxy).on_job_removed( # pyright: ignore[reportAny] on_job_removed )
try: expected_job = await cb
while True: job, result = await queue.get() if job == expected_job.path: return JobResult[result.upper()] finally: typing.cast('typing.Any', self.proxy).off_job_removed( # pyright: ignore[reportAny] on_job_removed )
@dataclassclass Systemd: bus: MessageBus manager: Manager
@staticmethod async def create(bus: MessageBus) -> 'Systemd': introspection = await bus.introspect(SYSTEMD_SERVICE, SYSTEMD_ROOT) systemd = bus.get_proxy_object(SYSTEMD_SERVICE, SYSTEMD_ROOT, introspection)
return Systemd(bus, Manager(proxy=systemd.get_interface(Manager.INTERFACE)))