Skip to content

Instantly share code, notes, and snippets.

@nvgoldin
Created April 23, 2017 16:09
Show Gist options
  • Save nvgoldin/75788c73313f32e20dfd5e1186aed8a0 to your computer and use it in GitHub Desktop.
Save nvgoldin/75788c73313f32e20dfd5e1186aed8a0 to your computer and use it in GitHub Desktop.
Running 'transient' systemd unit files with dbus
from __future__ import print_function
import uuid
import functools
import hashlib
# TO-DO: switch to pydbus, currently same code
# with pydbus gives errors.
from dbus import SessionBus, SystemBus, Interface
def rand_name():
return hashlib.sha256(str(uuid.uuid4())).hexdigest()[:8]
def get_systemd_manager(bus):
systemd = bus.get_object(
'org.freedesktop.systemd1', '/org/freedesktop/systemd1'
)
manager = Interface(
systemd, dbus_interface='org.freedesktop.systemd1.Manager'
)
return manager
def get_systemd_user():
return get_systemd_manager(SessionBus())
def get_systemd_system():
return get_systemd_manager(SystemBus())
def with_systemd_manager(scope='system'):
def wrap(func):
@functools.wraps(func)
def wrapped_func(*args, **kwargs):
if scope == 'system':
func(get_systemd_system(), *args, **kwargs)
elif scope == 'session':
func(get_systemd_user(), *args, **kwargs)
else:
raise RuntimeError('scope must be session or system')
return wrapped_func
return wrap
def start_transient(manager, cmd, name=None, mode='replace', props=None):
if cmd is None:
raise RuntimeError('no cmd was provided')
if name is None:
name = '{0}-{1}.service'.format('transient', rand_name())
if not isinstance(cmd, list):
raise RuntimeError('cmd must be a list')
if props is None:
properties = [
('Description', ' '.join(cmd)),
('ExecStart', [(cmd[0], cmd, False)])
]
else:
properties = props
job = manager.StartTransientUnit(name, mode, properties, [])
if not job:
raise RuntimeError('unable to start transient unit')
print('job started: {0}'.format(job))
print('service name: {0}'.format(name))
@with_systemd_manager(scope='session')
def start_user(manager, cmd, mode='replace'):
start_transient(manager=manager, cmd=cmd, mode=mode)
@with_systemd_manager(scope='system')
def start_system(manager, cmd, mode='replace'):
start_transient(manager=manager, cmd=cmd, mode=mode)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment