#!/usr/bin/python # # (c) 2008 Canonical Ltd. # Author: Martin Pitt # License: LGPL 2.1 or later # # Synchronous PackageKit client wrapper for Python. import gobject import dbus import dbus.mainloop.glib class PackageKitError(Exception): '''PackageKit error. This class mainly wraps a PackageKit "error enum". See http://www.packagekit.org/pk-reference.html#introduction-errors for details and possible values. ''' def __init__(self, error): self.error = error def __str__(self): return self.error class PackageKitClient: '''PackageKit client wrapper class. This exclusively uses synchonous calls. Functions which take a long time (install/remove packages) have callbacks for progress feedback. ''' def __init__(self): '''Initialize a PackageKit client.''' self.bus = dbus.SystemBus() self.pk_control = None self.main_loop = gobject.MainLoop() def SuggestDaemonQuit(self): '''Ask the PackageKit daemon to shutdown.''' try: self.pk_control.SuggestDaemonQuit() except (AttributeError, dbus.DBusException), e: # not initialized, or daemon timed out pass def Resolve(self, filter, package): '''Resolve a package name to a PackageKit package_id. filter and package are directly passed to the PackageKit transaction D-BUS method Resolve(). Return List of (installed, id, short_description) triples for all matches, where installed is a boolean and id and short_description are strings. ''' result = [] pk_xn = self._get_xn() pk_xn.connect_to_signal('Package', lambda i, p_id, summary: result.append((i == 'installed', str(p_id), str(summary)))) pk_xn.connect_to_signal('Finished', lambda status, code: self.main_loop.quit()) pk_xn.Resolve(filter, package) self.main_loop.run() return result def GetDetails(self, package_id): '''Get details about a PackageKit package_id. Return tuple (license, group, description, upstream_url, size). ''' result = [] pk_xn = self._get_xn() pk_xn.connect_to_signal('Details', lambda *args: result.extend(args)) pk_xn.connect_to_signal('Finished', lambda status, code: self.main_loop.quit()) pk_xn.GetDetails(package_id) self.main_loop.run() return (str(result[1]), str(result[2]), str(result[3]), str(result[4]), int(result[5])) def SearchName(self, filter, name): '''Search a package by name. Return a list of (installed, package_id, short_description) triples, where installed is a boolean and package_id/short_description are strings. ''' result = [] pk_xn = self._get_xn() pk_xn.connect_to_signal('Package', lambda i, id, summary: result.extend((str(id), str(summary)))) pk_xn.connect_to_signal('Finished', lambda status, code: self.main_loop.quit()) pk_xn.SearchName(filter, name) self.main_loop.run() return result def InstallPackages(self, package_ids, progress_cb=None): '''Install a list of package IDs. progress_cb is a function taking arguments (status, percentage, subpercentage, elapsed, remaining, allow_cancel). If it returns False, the action is cancelled (if allow_cancel == True), otherwise it continues. On failure this throws a PackageKitError or a DBusException. ''' self._InstRemovePackages(package_ids, progress_cb, True, None, None) def RemovePackages(self, package_ids, progress_cb=None, allow_deps=False, auto_remove=True): '''Remove a list of package IDs. progress_cb is a function taking arguments (status, percentage, subpercentage, elapsed, remaining, allow_cancel). If it returns False, the action is cancelled (if allow_cancel == True), otherwise it continues. allow_deps and auto_remove are passed to the PackageKit function. On failure this throws a PackageKitError or a DBusException. ''' self._InstRemovePackages(package_ids, progress_cb, False, allow_deps, auto_remove) # # Internal helper functions # def _InstRemovePackages(self, package_ids, progress_cb, install, allow_deps, auto_remove): '''Shared implementation of InstallPackages and RemovePackages.''' def _h_allowcancel(allow): global __InstallPackage_allow_cancel __InstallPackage_allow_cancel = allow def _h_status(status): global __InstallPackage_status __InstallPackage_status = status def _cancel(xn): try: xn.Cancel() except dbus.DBusException, e: if e._dbus_error_name == 'org.freedesktop.PackageKit.Transaction.CannotCancel': print 'cannot cancel' pass else: raise def _h_progress(per, subper, el, rem): global __InstallPackage_allow_cancel global __InstallPackage_status if __InstallPackage_status is not None: __InstallPackage_status = str(__InstallPackage_status) ret = progress_cb(__InstallPackage_status, int(per), int(subper), int(el), int(rem), __InstallPackage_allow_cancel == True) if not ret: gobject.timeout_add(10, _cancel, pk_xn) def _h_error(enum, desc): global __InstallPackage_error_enum __InstallPackage_error_enum = enum self.main_loop.quit() def _h_finished(status, code): global __InstallPackage_result __InstallPackage_result = status self.main_loop.quit() global __InstallPackage_error_enum, __InstallPackage_result global __InstallPackage_allow_cancel, __InstallPackage_status __InstallPackage_status = None __InstallPackage_error_enum = None __InstallPackage_result = None __InstallPackage_allow_cancel = False pk_xn = self._get_xn() if progress_cb: pk_xn.connect_to_signal('StatusChanged', _h_status) pk_xn.connect_to_signal('AllowCancel', _h_allowcancel) pk_xn.connect_to_signal('ProgressChanged', _h_progress) pk_xn.connect_to_signal('ErrorCode', _h_error) pk_xn.connect_to_signal('Finished', _h_finished) if install: pk_xn.InstallPackages(package_ids) else: pk_xn.RemovePackages(package_ids, allow_deps, auto_remove) self.main_loop.run() if __InstallPackage_error_enum: raise PackageKitError(__InstallPackage_error_enum) if __InstallPackage_result != 'success': raise PackageKitError('internal-error') def _get_xn(self): '''Create a new PackageKit Transaction object.''' try: tid = self.pk_control.GetTid() except (AttributeError, dbus.DBusException), e: if self.pk_control == None or (hasattr(e, '_dbus_error_name') and \ e._dbus_error_name == 'org.freedesktop.DBus.Error.ServiceUnknown'): # first initialization (lazy) or timeout self.pk_control = dbus.Interface(self.bus.get_object( 'org.freedesktop.PackageKit', '/org/freedesktop/PackageKit', False), 'org.freedesktop.PackageKit') tid = self.pk_control.GetTid() else: raise return dbus.Interface(self.bus.get_object('org.freedesktop.PackageKit', tid, False), 'org.freedesktop.PackageKit.Transaction') # # Test code # if __name__ == '__main__': import subprocess, sys dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) pk = PackageKitClient() print '---- Resolve() -----' print pk.Resolve('none', 'pmount') print pk.Resolve('none', 'quilt') print pk.Resolve('none', 'foobar') print '---- GetDetails() -----' print pk.GetDetails('installation-guide-powerpc;20080520ubuntu1;all;Ubuntu') print '---- SearchName() -----' print pk.SearchName('available', 'coreutils') print pk.SearchName('installed', 'coreutils') def cb(status, pc, spc, el, rem, c): print 'install pkg: %s, %i%%, cancel allowed: %s' % (status, pc, str(c)) return True #return pc < 12 print '---- InstallPackages() -----' pk.InstallPackages(['pmount;0.9.17-2;i386;Ubuntu', 'quilt;0.46-6;all;Ubuntu'], cb) subprocess.call(['dpkg', '-l', 'pmount', 'quilt']) print '---- RemovePackages() -----' pk.RemovePackages(['pmount;0.9.17-2;i386;Ubuntu', 'quilt;0.46-6;all;Ubuntu'], cb) subprocess.call(['dpkg', '-l', 'pmount', 'quilt']) pk.SuggestDaemonQuit()