Add python API, and convert pkt to using it
packagekit.py provides a simple PackageKit python API, along with the pkt client as an example user. The python API at the moment mirrors the actions supported by the APT backend (as that's mainly what I'm testing with), but should be expanded to be able to support the full set of actions (or at least the set supported by the user's backend of choice).
This commit is contained in:
Tom Parker
parent
f1c7a4f98a
commit
c273b0b90f
1
python/.gitignore
vendored
Normal file
1
python/.gitignore
vendored
Normal file
@ -0,0 +1 @@
|
||||
*.pyc
|
||||
147
python/packagekit.py
Normal file
147
python/packagekit.py
Normal file
@ -0,0 +1,147 @@
|
||||
# PackageKit python interface
|
||||
#
|
||||
# Copyright (C) 2007 Tom Parker <palfrey@tevp.net>
|
||||
#
|
||||
# Licensed under the GNU General Public License Version 2
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License v2 as published by
|
||||
# the Free Software Foundation.
|
||||
|
||||
from types import *
|
||||
import dbus
|
||||
from dbus.mainloop.glib import DBusGMainLoop
|
||||
import gobject
|
||||
|
||||
class PackageKitException(Exception):
|
||||
def __init__(self):
|
||||
Exception.__init__(self)
|
||||
|
||||
def __init__(self,e=None):
|
||||
Exception.__init__(self)
|
||||
if e == None:
|
||||
self._pk_name = None
|
||||
self._full_str = None
|
||||
else:
|
||||
if not isinstance(e,dbus.exceptions.DBusException):
|
||||
raise Exception,"Can only handle DBusExceptions"
|
||||
self._pk_name = str(e.get_dbus_name())
|
||||
self._full_str = str(e)
|
||||
|
||||
def get_backend_name(self):
|
||||
return self._pk_name
|
||||
|
||||
def __str__(self):
|
||||
if self._full_str!=None:
|
||||
return self._full_str
|
||||
else:
|
||||
return ""
|
||||
|
||||
class PackageKitNotStarted(PackageKitException):
|
||||
pass
|
||||
|
||||
class PackageKitAccessDenied(PackageKitException):
|
||||
pass
|
||||
|
||||
class PackageKitJobFailure(PackageKitException):
|
||||
pass
|
||||
|
||||
class PackageKitBackendFailure(PackageKitException):
|
||||
pass
|
||||
|
||||
def dbusException(func):
|
||||
def wrapper(*args,**kwargs):
|
||||
try:
|
||||
return func(*args,**kwargs)
|
||||
except dbus.exceptions.DBusException,e:
|
||||
if e.get_dbus_name() == "org.freedesktop.DBus.Error.AccessDenied":
|
||||
raise PackageKitAccessDenied
|
||||
elif e.get_dbus_name() == "org.freedesktop.DBus.Error.NoReply":
|
||||
raise PackageKitBackendFailure
|
||||
else:
|
||||
raise PackageKitException(e)
|
||||
except Exception:
|
||||
print "wibble"
|
||||
raise
|
||||
return wrapper
|
||||
|
||||
def job_id(func):
|
||||
def wrapper(*args,**kwargs):
|
||||
jid = func(*args,**kwargs)
|
||||
if jid == -1:
|
||||
raise PackageKitJobFailure
|
||||
else:
|
||||
return jid
|
||||
return wrapper
|
||||
|
||||
class PackageKit:
|
||||
def __init__(self):
|
||||
DBusGMainLoop(set_as_default=True)
|
||||
bus = dbus.SystemBus()
|
||||
try:
|
||||
pk = bus.get_object('org.freedesktop.PackageKit', '/org/freedesktop/PackageKit')
|
||||
self.pk_iface = dbus.Interface(pk, dbus_interface='org.freedesktop.PackageKit')
|
||||
except dbus.exceptions.DBusException,e:
|
||||
if e.get_dbus_name() == "org.freedesktop.DBus.Error.ServiceUnknown":
|
||||
raise PackageKitNotStarted
|
||||
else:
|
||||
raise PackageKitException(e)
|
||||
|
||||
self.job = None
|
||||
self.progress = 0.0
|
||||
bus.add_signal_receiver(self.catchall_signal_handler, interface_keyword='dbus_interface', member_keyword='member',dbus_interface="org.freedesktop.PackageKit")
|
||||
|
||||
def run(self):
|
||||
self.loop = gobject.MainLoop()
|
||||
self.loop.run()
|
||||
|
||||
def catchall_signal_handler(self,*args, **kwargs):
|
||||
if args[0] != self.job and kwargs['member']!="JobListChanged":
|
||||
return
|
||||
if kwargs['member'] == "Finished":
|
||||
loop.quit()
|
||||
self.Finish()
|
||||
elif kwargs['member'] == "PercentageChanged":
|
||||
progress = float(args[1])+(progress%1.0)
|
||||
self.Percentage(progress)
|
||||
elif kwargs['member'] == "SubPercentageChanged":
|
||||
progress = (float(args[1])/100.0)+int(progress)
|
||||
self.Percentage(progress)
|
||||
elif kwargs['member'] == "JobStatusChanged":
|
||||
self.JobStatus(args[1])
|
||||
elif kwargs['member'] == "Package":
|
||||
self.Package(args[2],args[3])
|
||||
elif kwargs['member'] in ["NoPercentageUpdates","JobListChanged"]:
|
||||
pass
|
||||
else:
|
||||
print "Caught signal %s"% kwargs['member']
|
||||
for arg in args:
|
||||
print " " + str(arg)
|
||||
|
||||
def Percentage(self,value):
|
||||
pass
|
||||
|
||||
def JobStatus(self,string):
|
||||
pass
|
||||
|
||||
def Finish(self):
|
||||
pass
|
||||
|
||||
@dbusException
|
||||
@job_id
|
||||
def SearchName(self,pattern,filter="none"):
|
||||
return self.pk_iface.SearchName(filter,pattern)
|
||||
|
||||
@dbusException
|
||||
@job_id
|
||||
def GetDescription(package_id):
|
||||
return pk_iface.GetDescription(package_id)
|
||||
|
||||
@dbusException
|
||||
@job_id
|
||||
def RefreshCache(force=False):
|
||||
return pk_iface.RefreshCache(force)
|
||||
|
||||
# hack to avoid exporting them
|
||||
del job_id
|
||||
del dbusException
|
||||
83
python/pkt
Executable file
83
python/pkt
Executable file
@ -0,0 +1,83 @@
|
||||
#!/usr/bin/python
|
||||
#
|
||||
# "pkt" python test program for PackageKit
|
||||
#
|
||||
# pkt serves both as a simple PackageKit client, and as an example user of the
|
||||
# PackageKit python API
|
||||
#
|
||||
# Copyright (C) 2007 Tom Parker <palfrey@tevp.net>
|
||||
#
|
||||
# Licensed under the GNU General Public License Version 2
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License v2 as published by
|
||||
# the Free Software Foundation.
|
||||
|
||||
from sys import argv,exit
|
||||
from optparse import OptionParser
|
||||
from types import FunctionType
|
||||
|
||||
from packagekit import *
|
||||
|
||||
class pkt(PackageKit):
|
||||
def Percentage(self,progress):
|
||||
print "Progress: %.2f%%"%progress
|
||||
|
||||
def JobStatus(self,type):
|
||||
print "Job type: %s"%type
|
||||
|
||||
def Package(self,name,description):
|
||||
print "Package: %s - %s"%(name,description)
|
||||
|
||||
try:
|
||||
p = pkt()
|
||||
except PackageKitNotStarted:
|
||||
print "PackageKit doesn't appear to be started. You may need to enable dbus autostart"
|
||||
exit(1)
|
||||
|
||||
def search(*args):
|
||||
patt = " ".join(args[0])
|
||||
return p.SearchName(patt)
|
||||
|
||||
def desc(*args):
|
||||
if len(args)!=1 or len(args[0])!=1:
|
||||
print "desc only takes single arg"
|
||||
return -1
|
||||
return p.GetDescription(args[0][0])
|
||||
|
||||
def update(args):
|
||||
if len(args)>0 and len(args[0])>0:
|
||||
print "update doesn't take args"
|
||||
return -1
|
||||
return p.RefreshCache()
|
||||
|
||||
def usage():
|
||||
print "Usage: %s <command> <options>"%argv[0]
|
||||
print "Valid commands are:"
|
||||
for k in globals().keys():
|
||||
if k in ["usage","catchall_signal_handler"]: #ignore
|
||||
continue
|
||||
g = globals()[k]
|
||||
if type(g) == FunctionType:
|
||||
print "\t%s"%k
|
||||
exit(1)
|
||||
|
||||
parser = OptionParser()
|
||||
(options, args) = parser.parse_args()
|
||||
|
||||
if len(args)==0:
|
||||
usage()
|
||||
|
||||
if not globals().has_key(args[0]):
|
||||
print "Don't know operation '%s'"%args[0]
|
||||
usage()
|
||||
|
||||
try:
|
||||
job = globals()[args[0]](args[1:])
|
||||
except PackageKitAccessDenied:
|
||||
print "You don't have sufficient permissions to access PackageKit (on the org.freedesktop.PackageKit dbus service)"
|
||||
exit(1)
|
||||
except PackageKitJobFailure:
|
||||
usage()
|
||||
|
||||
p.run()
|
||||
109
tools/pkt
109
tools/pkt
@ -1,109 +0,0 @@
|
||||
#!/usr/bin/python
|
||||
#
|
||||
# Copyright (C) 2007 Tom Parker <palfrey@tevp.net>
|
||||
#
|
||||
# Licensed under the GNU General Public License Version 2
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License v2 as published by
|
||||
# the Free Software Foundation.
|
||||
|
||||
from sys import argv,exit
|
||||
from types import *
|
||||
import dbus
|
||||
from dbus.mainloop.glib import DBusGMainLoop
|
||||
DBusGMainLoop(set_as_default=True)
|
||||
import gobject
|
||||
from optparse import OptionParser
|
||||
|
||||
bus = dbus.SystemBus()
|
||||
try:
|
||||
pk = bus.get_object('org.freedesktop.PackageKit', '/org/freedesktop/PackageKit')
|
||||
pk_iface = dbus.Interface(pk, dbus_interface='org.freedesktop.PackageKit')
|
||||
except dbus.exceptions.DBusException,e:
|
||||
if e.get_dbus_name() == "org.freedesktop.DBus.Error.ServiceUnknown":
|
||||
print "PackageKit doesn't appear to be started. You may need to enable dbus autostart"
|
||||
exit(1)
|
||||
else:
|
||||
raise
|
||||
|
||||
job = None
|
||||
progress = 0.0
|
||||
|
||||
def catchall_signal_handler(*args, **kwargs):
|
||||
global job,progress
|
||||
if args[0] != job and kwargs['member']!="JobListChanged":
|
||||
print args,job,kwargs
|
||||
raise Exception
|
||||
if kwargs['member'] == "Finished":
|
||||
loop.quit()
|
||||
elif kwargs['member'] == "PercentageChanged":
|
||||
progress = float(args[1])+(progress%1.0)
|
||||
print "Progress: %.2f%%"%progress
|
||||
elif kwargs['member'] == "SubPercentageChanged":
|
||||
progress = (float(args[1])/100.0)+int(progress)
|
||||
print "Progress: %.2f%%"%progress
|
||||
elif kwargs['member'] == "JobStatusChanged":
|
||||
print "Job type: %s"%args[1]
|
||||
elif kwargs['member'] == "Package":
|
||||
print "Package: %s - %s"%(args[2],args[3])
|
||||
elif kwargs['member'] in ["NoPercentageUpdates","JobListChanged"]:
|
||||
pass
|
||||
else:
|
||||
print "Caught signal %s"% kwargs['member']
|
||||
for arg in args:
|
||||
print " " + str(arg)
|
||||
|
||||
bus.add_signal_receiver(catchall_signal_handler, interface_keyword='dbus_interface', member_keyword='member',dbus_interface="org.freedesktop.PackageKit")
|
||||
|
||||
def search(*args):
|
||||
patt = " ".join(args[0])
|
||||
return pk_iface.SearchName("none",patt)
|
||||
|
||||
def desc(*args):
|
||||
if len(args)!=1 or len(args[0])!=1:
|
||||
print "desc only takes single arg"
|
||||
return -1
|
||||
return pk_iface.GetDescription(args[0][0])
|
||||
|
||||
def update(args):
|
||||
if len(args)>0 and len(args[0])>0:
|
||||
print "update doesn't take args"
|
||||
return -1
|
||||
return pk_iface.RefreshCache(False)
|
||||
|
||||
def usage():
|
||||
print "Usage: %s <command> <options>"%argv[0]
|
||||
print "Valid commands are:"
|
||||
for k in globals().keys():
|
||||
if k in ["usage","catchall_signal_handler"]: #ignore
|
||||
continue
|
||||
g = globals()[k]
|
||||
if type(g) == FunctionType:
|
||||
print "\t%s"%k
|
||||
exit(1)
|
||||
|
||||
parser = OptionParser()
|
||||
(options, args) = parser.parse_args()
|
||||
|
||||
if len(args)==0:
|
||||
usage()
|
||||
|
||||
if not globals().has_key(args[0]):
|
||||
print "Don't know operation '%s'"%args[0]
|
||||
usage()
|
||||
|
||||
try:
|
||||
job = globals()[args[0]](args[1:])
|
||||
except dbus.exceptions.DBusException,e:
|
||||
if e.get_dbus_name() == "org.freedesktop.DBus.Error.AccessDenied":
|
||||
print "You don't have sufficient permissions to access PackageKit (on the org.freedesktop.PackageKit dbus service)"
|
||||
exit(1)
|
||||
else:
|
||||
raise
|
||||
|
||||
if job == -1:
|
||||
usage()
|
||||
|
||||
loop = gobject.MainLoop()
|
||||
loop.run()
|
||||
Loading…
Reference in New Issue
Block a user