Attachment 427 Details for Bug 4208 – packagekit prototype

packagekit prototype

pkginstaller.py (text/x-python), 11.05 KB, created by Henrik Andersson on 2012-02-27 14:24:20 CET

(hide)

Description:

Filename:

MIME Type:

Creator: Henrik Andersson

Created: 2012-02-27 14:24:20 CET

Size: 11.05 KB

patch

obsolete

>#!/usr/bin/python
>
>import gtk
>import gobject
>import dbus
>
>import dbus.mainloop.glib
>dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
>
>from dbus import DBusException
>
>def test_packagekit():
>    b = dbus.SystemBus()
>    ipk = dbus.Interface(b.get_object(
>            "org.freedesktop.PackageKit",
>            "/org/freedesktop/PackageKit",
>            False), "org.freedesktop.PackageKit")
>
>#### PolicyKit authentication borrowed wrapper ##
>class PermissionDeniedByPolicy(dbus.DBusException):
>    _dbus_error_name = 'org.freedesktop.PackageKit.Transaction.RefusedByPolicy'
>
>def polkit_auth_wrapper(fn, *args, **kwargs):
>    '''Function call wrapper for PolicyKit authentication.
>
>    Call fn(*args, **kwargs). If it fails with a PermissionDeniedByPolicy
>    and the caller can authenticate to get the missing privilege, the PolicyKit
>    authentication agent is called, and the function call is attempted again.
>    '''
>    try:
>        return fn(*args, **kwargs)
>    except dbus.DBusException, e:
>        if e._dbus_error_name == PermissionDeniedByPolicy._dbus_error_name:
>            # last words in message are privilege and auth result
>            (priv, auth_result) = e.message.split()[-2:]
>            if auth_result.startswith('auth_'):
>                pk_auth = dbus.SessionBus().get_object(
>                    'org.freedesktop.PolicyKit.AuthenticationAgent', '/',
>                    'org.gnome.PolicyKit.AuthorizationManager.SingleInstance')
>                # TODO: provide xid
>                res = pk_auth.ObtainAuthorization(priv, dbus.UInt32(0),
>                    dbus.UInt32(os.getpid()), timeout=300)
>                print res
>                if res:
>                    return fn(*args, **kwargs)
>            raise PermissionDeniedByPolicy(priv + ' ' + auth_result)
>        else:
>            raise
>
>
>class PackageKitPackage:
>    """ a packagekit package constructed from id string """
>    def __init__(self, id):
>      self.__id = id
>      self.installed = False
>      
>      if id.find("installed") != -1:
>          self.installed = True
>
>    def __str__(self):
>        return self.__id
>
>    def name(self):
>        """ get a pretty name for the package """
>        (name, ver, arch, repo) = tuple(self.__id.split(";"))
>        return ("%s-%s.%s" % (name, ver, arch))
>
>class PackageKitTransaction:
>    """ a packagekit transaction carried out using dbus packagekit service 
>        see, http://www.packagekit.org/gtk-doc/PackageKit/Transaction.html for dbus docs.
>    """
>
>    __transaction_id = ""
>
>    def __init__(self, frontend, method, *args):
>        self.__frontend = frontend
>        self.__method = method
>        self.__args = args
>        self.__main_loop = gobject.MainLoop()
>        self.__error = 0
>        self.__error_message = ""
>
>        self.packages = []
>
>    def run(self, blocking = False):
>        """ setup and run transaction """
>        if self.__transaction_id:
>            return
>	
>        # initialize and carry out the transaction
>        bus = dbus.SystemBus()
>      
>        ipkgkit = dbus.Interface(bus.get_object(
>                        'org.freedesktop.PackageKit',
>                        '/org/freedesktop/PackageKit',
>                    False), 'org.freedesktop.PackageKit')
>
>        self.__transaction_id = ipkgkit.GetTid()
>
>        itrans = dbus.Interface(bus.get_object('org.freedesktop.PackageKit', self.__transaction_id, False),
>                                'org.freedesktop.PackageKit.Transaction')
>
>        # connect signals for a transaction 
>        for signal, callback in [
>            ('Finished', self.__on_finished),
>            ('Package', self.__on_package),
>            ('Message', self.__on_message)
>            ]:
>            itrans.connect_to_signal(signal, callback)
>
>                
>        method = itrans.get_dbus_method(self.__method)
>
>        # carry out the method call
>        context = gobject.main_context_default()
>        while context.pending():
>            context.iteration()
>
>	polkit_auth_wrapper(method, *self.__args)
>
>        #//method(*self.__args)
>
>        if blocking:
>            self.__main_loop.run()
>        
>
>    def rollback(self):
>        """ rollback transaction """
>        if not self.__transaction_id:
>            return
>
>    def error(self):
>        if self.__error:
>            return self.__error_message
>        return None
>
>    # Callback dispatchers for frontend
>    def __on_finished(self, exit, runtime):
>       self.__frontend._on_finished()
>       self.__main_loop.quit()
>
>    def __on_package(self, i, id, summary):
>        self.packages.append(PackageKitPackage(id))
>
>    def __on_message(self,type, details):
>        self.__frontend._on_message(type, details)
>        
>
>class PackageKitFrontend:
>    def __init__(self):
>        test_packagekit()
>
>    def transaction(self, method, *args):
>        """ creates a new transaction """
>        return PackageKitTransaction(self, method, *args)
>
>    def _on_finished(self):
>        """ callback for transaction finished """
>        pass
>
>    def _on_message(self, type, message):
>        """ callback for a transaction message """
>        pass
>
>
>class PackageInstaller(PackageKitFrontend):
>    """The pkgkit installer helper class"""
>    packages = []
>    depends = []
>    transactions = {}
>
>    def __init__(self):
>        PackageKitFrontend.__init__(self)
>
>    def add_package(self, package, filter="none"):
>        """Queue up a package that is to be installed"""
>        # check if package is installed, note resolve
>        # can return a list of matched package by name
>        t = PackageKitFrontend.transaction(self, "Resolve", filter + ";~installed", [package])
>        t.run(True)
>
>	if not t.packages:
>	    return []
>
>        if t.error():
>            print "failed to resolve package by name '" + package +"'."
>            return []
>	
>	for p in t.packages:
>	    self.packages.append(p)
>
>	self.__resolve_depends(t.packages, filter)
>
>    def __resolve_depends(self, packages, filter):
>        for p in packages:
>            if not p.installed:
>                # resolve dependencies for specific package
>                td = PackageKitFrontend.transaction(self, "GetDepends", filter + ";~installed", [str(p) for p in packages], True)
>                td.run(True)
>		if not td.packages:
>		    return []
>
>                if td.error():
>                    print "failed to fetch list of packages depends"
>                    return []
>
>                for d in td.packages:
>                    if not d.installed:
>                        self.depends.append(d)
>                
>
>    def add_package_for_file(self, file, filter="none"):
>	"""Queue up a package that owns a file to be installed"""
>	t = PackageKitFrontend.transaction(self, "SearchFiles", filter + ";~installed", [file])
>	t.run(True)
>	
>	if not t.packages:
>	    return []
>
>	if t.error():
>	    print "failed to find package by file '" + file + "'."
>	    return []
>
>	for p in t.packages:
>	    self.packages.append(p)
>
>	self.__resolve_depends(t.packages, filter) 
>
>
>    def install(self):
>        t = PackageKitFrontend.transaction(self, "InstallPackages", True,
>                                       [ str(p) for p in self.packages])
>        t.run(False)
>
>    def on_message(self):
>        pass
>
>    def on_finished(self):
>        pass
>
>    # called when all transactions are finished
>    def _on_finished(self):
>        self.on_finished()
>
>    def _on_message(self, type, details):
>        print "onMessage: " + str(type) +" message: " + details 
>
>
>class PackageInstallerGui(PackageInstaller):
>    def __init__(self):
>        PackageInstaller.__init__(self)
>        self.notebook = gtk.Notebook()
>        self.notebook.set_border_width(10)
>        self.notebook.set_show_border(False)
>
>        # initialize the info page
>        self.info_page = gtk.VBox()
>        self.infolabel = gtk.Label("Following packages need to be installed:")
>        self.infolabel.set_alignment(0,0.5)
>        self.info_page.pack_start(self.infolabel, False, False, 2)
>        self.notebook.add(self.info_page)
>        self.notebook.set_show_tabs(False)
>
>        sw = gtk.ScrolledWindow()
>        sw.set_size_request(-1, 128)
>        self.info_page.pack_start(sw, False, True, 2)
>        self.pkgs_store = gtk.ListStore(gobject.TYPE_STRING)
>        self.pkgs_view = gtk.IconView(self.pkgs_store)
>        self.pkgs_view.set_text_column(0)
>        #self.pkgs_view.set_item_padding(2)
>        sw.add_with_viewport(self.pkgs_view)
>
>        self.depslabel = gtk.Label("which will install/update the following packages:")
>        self.depslabel.set_alignment(0,0.5)
>        self.info_page.pack_start(self.depslabel, False, False, 2)
>
>        sw = gtk.ScrolledWindow()
>        self.info_page.pack_start(sw, True, True, 2)
>        self.deps_store = gtk.ListStore(gobject.TYPE_STRING)
>        self.deps_view = gtk.IconView(self.deps_store)
>        self.deps_view.set_text_column(0)
>        #self.deps_view.set_item_padding(2)
>        sw.add_with_viewport(self.deps_view)
>
>        # initialize the progress page
>        self.installation_page = gtk.VBox()
>        self.notebook.add(self.installation_page)
>        self.installation_page.pack_start(gtk.ProgressBar())
>
>        # initialize the result page
>        self.result_page = gtk.VBox()
>        self.notebook.add(self.result_page)
>        self.result_page.pack_start(gtk.Label("The installation result page"))
>
>    def initialize(self):
>        # initialize the info page
>        self.depslabel.set_text("which will install/update the following " + str(len(PackageInstaller.depends)) + " packages:")
>
>        # add packages to listviews
>        if PackageInstaller.packages:
>            for p in PackageInstaller.packages:
>                self.pkgs_store.append([p.name()])
>
>        if PackageInstaller.depends:
>            for p in PackageInstaller.depends:
>                self.deps_store.append([p.name()])
>
>        return self.notebook
>
>    def on_finished(self):
>        # all transaction are finished, lets step to summary page
>        self.next()
>
>    def next(self):
>        if (self.notebook.get_current_page() == self.notebook.get_n_pages()-1):
>            return False
>
>        self.notebook.next_page()
>
>        if (self.notebook.get_current_page() == 1):
>            self.install()
>            return True
>
>        if (self.notebook.get_current_page() == 2):
>            self.summary()
>            return True
>       
>    def install(self):
>        PackageInstaller.install(self)
>        pass
>
>    def summary(self):
>        pass
>
>class PackageInstallerConsole(PackageInstaller):
>    def __init__(self):
>        pass
>
>def mydestroy(widget, data=None):
>    gtk.main_quit()
>
>def next(widget, data=None):
>   if (data.next() == False):
>       gtk.main_quit()
>
># create a assitant and add package installer pages
>pi = PackageInstallerGui()
>pi.add_package("dialog","arch")
>#pi.add_package_for_file("/usr/bin/X","arch")
>#pi.add_package("digikam","arch")
>#pi.add_package("kobby","arch")
>
>w = gtk.Window()
>w.set_size_request(640,480)
>w.connect("destroy", mydestroy)
>c = gtk.VBox(False)
>w.add(c)
>
># add content page
>c.pack_start(pi.initialize(),True, True, 2)
>
># add button row
>br = gtk.HBox(False)
>c.pack_start(br, False,False,2)
>
>b = gtk.Button("next")
>b.connect("clicked", next, pi)
>br.pack_end(b, False,False,2)
>
>w.show_all()
>
>gtk.main()
>
>

Actions: View

Attachments on bug 4208: 427