Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
Handle call result.
:param Gio.DBusProxy proxy:
:param Gio.AsyncResult result:
:param user_data: unused
"""
try:
value = proxy.call_finish(result)
except Exception as e:
self.errback(e, format_exc())
else:
self.callback(*unpack_variant(value))
class DBusCallWithFdList(Async):
"""
Asynchronously call a DBus method.
"""
def __init__(self,
proxy,
method_name,
signature,
args,
fds,
flags=0,
timeout_msec=-1):
"""
Asynchronously call the specified method on a DBus proxy object.
event,
object_path,
None,
Gio.DBusSignalFlags.NONE,
callback,
None,
)
def disconnect(self, subscription_id):
"""
Disconnect a DBus signal subscription.
"""
self.connection.signal_unsubscribe(subscription_id)
class DBusProxyNew(Async):
"""
Asynchronously call a DBus method.
"""
def __init__(self,
connection,
flags,
info,
name,
object_path,
interface_name):
"""
Asynchronously call the specified method on a DBus proxy object.
"""
super(DBusProxyNew, self).__init__()
:param Gio.DBusProxy proxy:
:param Gio.AsyncResult result:
:param user_data: unused
"""
try:
value = Gio.DBusProxy.new_finish(result)
if value is None:
raise RuntimeError("Failed to connect DBus object!")
except Exception as e:
self.errback(e, format_exc())
else:
self.callback(value)
class DBusProxyNewForBus(Async):
"""
Asynchronously call a DBus method.
"""
def __init__(self,
bus_type,
flags,
info,
name,
object_path,
interface_name):
"""
Asynchronously call the specified method on a DBus proxy object.
"""
super(DBusProxyNewForBus, self).__init__()
def __init__(self, menumaker, _icons):
self._maker = menumaker
self._menu = Gtk.Menu()
self._indicator = AppIndicator3.Indicator.new(
'udiskie',
_icons.get_icon_name('media'),
AppIndicator3.IndicatorCategory.HARDWARE)
self._indicator.set_status(AppIndicator3.IndicatorStatus.PASSIVE)
self._indicator.set_menu(self._menu)
# Get notified before menu is shown, see:
# https://bugs.launchpad.net/screenlets/+bug/522152/comments/15
dbusmenuserver = self._indicator.get_property('dbus-menu-server')
self._dbusmenuitem = dbusmenuserver.get_property('root-node')
self._conn = self._dbusmenuitem.connect('about-to-show', self._on_show)
self.task = Async()
menumaker._quit_action = self.destroy
# Populate menu initially, so libdbusmenu does not ignore the
# 'about-to-show':
self._maker(self._menu)
return dict(map(unpack_variant, elems))
if t.is_tuple():
return tuple(map(unpack_variant, elems))
if t.is_array():
return list(map(unpack_variant, elems))
if t.is_dict_entry():
return list(map(unpack_variant, elems))
# The following cases should never occur for DBus data:
if t.is_variant():
return unpack_variant(elems[0])
if t.is_maybe():
return unpack_variant(elems[0]) if elems else None
raise ValueError("Unknown variant type: {}!".format(v.get_type_string()))
class DBusCall(Async):
"""
Asynchronously call a DBus method.
"""
def __init__(self,
proxy,
method_name,
signature,
args,
flags=0,
timeout_msec=-1):
"""
Asynchronously call the specified method on a DBus proxy object.
:param Gio.DBusProxy proxy: