Skip to content

Commit

Permalink
Add & implement org.gnome.GTG.Tasks DBus interface
Browse files Browse the repository at this point in the history
Introducing org.gnome.GTG.Tasks, an DBus interface to manage tasks. This
is very much inspired by the old "DBus Wrapper", that implemented
org.gnome.GTGTasks (I think), but this is in some ways different:

* Do actions in batches
* Only handle tasks, no UI stuff (actions can replace some)
  • Loading branch information
Neui committed Mar 26, 2021
1 parent 93989df commit 23a6d30
Show file tree
Hide file tree
Showing 8 changed files with 529 additions and 27 deletions.
76 changes: 49 additions & 27 deletions GTG/core/dbus/dbus.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,13 @@
from gi.repository import GLib, Gio
from GTG.core.logger import log
import logging
import os
import traceback


def _get_installed_xml_from_interface_name(interface_name):
log = logging.getLogger(__name__)


def _get_installed_xml_from_interface_name(interface_name: str):
"""Return an DBus interface XML from system installed folders"""
raw_paths = GLib.get_system_data_dirs()
raw_paths.append(GLib.get_user_data_dir())
Expand All @@ -18,14 +21,20 @@ def _get_installed_xml_from_interface_name(interface_name):
pass
return None

def _get_internal_xml_from_interface_name(interface_name):

def _get_internal_xml_from_interface_name(interface_name: str):
"""Return an DBus interface XML from internal data folders"""
log.debug("TODO: Implement this later")
pass

def get_xml_from_interface_name(interface_name, use_internal=True, use_system=True):
"""Return an DBus interface XML from either provided by the system (distro)
or internal"""

def get_xml_from_interface_name(interface_name: str,
use_internal: bool = True,
use_system: bool = True):
"""
Return an DBus interface XML from either provided by the system (distro)
or internal.
"""
if use_internal:
xml = _get_internal_xml_from_interface_name(interface_name)
if xml is not None:
Expand All @@ -38,13 +47,16 @@ def get_xml_from_interface_name(interface_name, use_internal=True, use_system=Tr


class DBusReturnError(Exception):
"""An error that an DBus interface implementation can throw to indicate
that it should return with that error"""
"""
An error that an DBus interface implementation can throw to indicate
that it should return with that error.
"""
def __init__(self, name, message):
super().__init__(message)
self.name = str(name)
self.message = str(message)


class DBusInterfaceImplService():
INTERFACE_NAME = None
NODE_INFO = None
Expand Down Expand Up @@ -76,61 +88,71 @@ def _get_info(self):
log.debug(f"Got interface info for {self.INTERFACE_NAME}: {self.INTERFACE_INFO}")

def dbus_register(self, dbus_connection, object_path):
"""Register this implementation on the specified connection and object"""
"""
Register this implementation on the specified connection and object.
"""
self._get_info()
self._object_id = dbus_connection.register_object(object_path,
self.INTERFACE_INFO,
self._handle_method_call,
self._handle_get_property, self._handle_set_property)
self._object_id = dbus_connection.register_object(
object_path,
self.INTERFACE_INFO,
self._handle_method_call,
self._handle_get_property, self._handle_set_property)
self._dbus_connection = dbus_connection
return self._object_id

def dbus_unregister(self):
"""Unregister this implementation"""
"""Unregister this implementation."""
if self._dbus_connection is not None and self._object_id is not None:
log.debug(f"Unregister {self.INTERFACE_NAME} on {self._dbus_connection} via {self._object_id}")
log.debug("Unregister %r on %r via %r",
self.INTERFACE_NAME,
self._dbus_connection,
self._object_id)
ret = self._dbus_connection.unregister_object(self._object_id)
self._dbus_connection = None
self._object_id = None
if not ret:
log.warn(f"Unregister for {self.INTERFACE_NAME} failed!")
log.warn("Unregister for %r failed!", self.INTERFACE_NAME)
return ret
else:
log.warn(f"Trying to unregister not-registered {self.INTERFACE_NAME}")
log.warn("Trying to unregister not-registered %r",
self.INTERFACE_NAME)
return False

def _handle_method_call(self, connection,
sender, object_path,
interface_name, method_name,
parameters, invocation):
dbg = f"{interface_name}.{method_name}{parameters} on {object_path} by {sender}"
log.debug(f"Called {dbg}")
sender, object_path,
interface_name, method_name,
parameters, invocation):
dbg = "%s.%s%s on %s by %s"
dbg_args = (interface_name, method_name, parameters,
object_path, sender)
log.debug(f"Called {dbg}", *dbg_args)

try:
method = self.__getattribute__(method_name)
except AttributeError as e:
log.debug(f"{dbg} → Internal python exception: {str(e)}")
log.debug(f"{dbg} → Internal python exception: %s", *dbg_args, e)
invocation.return_dbus_error("python." + type(e).__name__, str(e))
return

try:
ret = method(*parameters.unpack())
except GLib.Error as e:
log.debug(f"{dbg} → GLib Error: {str(e)}")
log.debug(f"{dbg} → GLib Error: %s", *dbg_args, e)
invocation.return_gerror(e)
except DBusReturnError as e:
log.debug(f"{dbg} → Custom Error: {e.name, e.message}")
log.debug(f"{dbg} → Custom Error: %s: %s",
*dbg_args, e.name, e.message)
invocation.return_dbus_error(e.name, e.message)
except Exception as e:
log.warn(f"{dbg} → Python exception: {str(e)}")
log.warn(f"{dbg} → Python exception: %s", *dbg_args, e)
traceback.print_exc()
invocation.return_dbus_error("python." + type(e).__name__, str(e))
else:
if type(ret) != GLib.Variant:
if len(invocation.get_method_info().out_args) == 1:
ret = (ret,) # Make one-value return work as expected
ret = GLib.Variant(self.__get_return_variant(invocation), ret)
log.debug(f"{dbg} → Returning {ret}")
log.debug(f"{dbg} → Returning %r", *dbg_args, ret)
invocation.return_value(ret)

def _handle_get_property(self, *args):
Expand Down
Loading

0 comments on commit 23a6d30

Please sign in to comment.