Auto-throw exceptions when resuming tasks
[zeroinstall.git] / zeroinstall / injector / packagekit.py
blob0aa29f7b87142492c2ddda1e5b710a2d50788744
1 """
2 PackageKit integration.
3 """
5 # Copyright (C) 2010, Aleksey Lim
6 # See the README file for details, or visit http://0install.net.
8 import os, sys
9 import locale
10 import logging
11 from zeroinstall import _, SafeException
13 from zeroinstall.support import tasks
14 from zeroinstall.injector import download, model
16 _logger_pk = logging.getLogger('packagekit')
18 try:
19 import dbus
20 import dbus.mainloop.glib
21 dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
22 except Exception as ex:
23 _logger_pk.info("D-BUS not available: %s", ex)
24 dbus = None
26 MAX_PACKAGE_KIT_TRANSACTION_SIZE = 100
28 class PackageKit(object):
29 def __init__(self):
30 self._pk = False
32 self._candidates = {} # { package_name : [ (version, arch, size) ] | Blocker }
34 # PackageKit is really slow at handling separate queries, so we use this to
35 # batch them up.
36 self._next_batch = set()
38 @property
39 def available(self):
40 return self.pk is not None
42 @property
43 def pk(self):
44 if self._pk is False:
45 try:
46 self._pk = dbus.Interface(dbus.SystemBus().get_object(
47 'org.freedesktop.PackageKit',
48 '/org/freedesktop/PackageKit', False),
49 'org.freedesktop.PackageKit')
50 _logger_pk.info(_('PackageKit dbus service found'))
51 except Exception as ex:
52 _logger_pk.info(_('PackageKit dbus service not found: %s'), ex)
53 self._pk = None
54 return self._pk
56 def get_candidates(self, package_name, factory, prefix):
57 """Add any cached candidates.
58 The candidates are those discovered by a previous call to L{fetch_candidates}.
59 @param package_name: the distribution's name for the package
60 @param factory: a function to add a new implementation to the feed
61 @param prefix: the prefix for the implementation's ID
62 """
63 candidates = self._candidates.get(package_name, None)
64 if candidates is None:
65 return
67 if isinstance(candidates, tasks.Blocker):
68 return # Fetch still in progress
70 for candidate in candidates:
71 impl_name = '%s:%s:%s:%s' % (prefix, package_name, candidate['version'], candidate['arch'])
73 impl = factory(impl_name, only_if_missing = True, installed = candidate['installed'])
74 if impl is None:
75 # (checking this way because the cached candidate['installed'] may be stale)
76 return # Already installed
78 impl.version = model.parse_version(candidate['version'])
79 if candidate['arch'] != '*':
80 impl.machine = candidate['arch']
82 def install(handler):
83 packagekit_id = candidate['packagekit_id']
84 dl = PackageKitDownload('packagekit:' + packagekit_id, hint = impl, pk = self.pk, packagekit_id = packagekit_id, expected_size = candidate['size'])
85 handler.monitor_download(dl)
86 return dl.downloaded
87 impl.download_sources.append(model.DistributionSource(package_name, candidate['size'], install))
89 @tasks.async
90 def fetch_candidates(self, package_names):
91 assert self.pk
93 # Batch requests up
94 self._next_batch |= set(package_names)
95 yield
96 package_names = self._next_batch
97 self._next_batch = set()
98 # The first fetch_candidates instance will now have all the packages
99 # For the others, package_names will now be empty
101 known = [self._candidates[p] for p in package_names if p in self._candidates]
102 # (use set because a single task may be checking multiple packages and we need
103 # to avoid duplicates).
104 in_progress = list(set([b for b in known if isinstance(b, tasks.Blocker)]))
105 _logger_pk.debug('Already downloading: %s', in_progress)
107 # Filter out the ones we've already fetched
108 package_names = [p for p in package_names if p not in self._candidates]
110 def do_batch(package_names):
111 #_logger_pk.info("sending %d packages in batch", len(package_names))
112 versions = {}
114 blocker = None
116 def error_cb(sender):
117 # Note: probably just means the package wasn't found
118 _logger_pk.info(_('Transaction failed: %s(%s)'), sender.error_code, sender.error_details)
119 blocker.trigger()
121 def details_cb(sender):
122 for packagekit_id, info in versions.items():
123 if packagekit_id in sender.details:
124 info.update(sender.details[packagekit_id])
125 info['packagekit_id'] = packagekit_id
126 if (info['name'] not in self._candidates or
127 isinstance(self._candidates[info['name']], tasks.Blocker)):
128 self._candidates[info['name']] = [info]
129 else:
130 self._candidates[info['name']].append(info)
131 else:
132 _logger_pk.info(_('Empty details for %s'), packagekit_id)
133 blocker.trigger()
135 def resolve_cb(sender):
136 if sender.package:
137 for packagekit_id, info in sender.package.iteritems():
138 parts = packagekit_id.split(';', 3)
139 if ':' in parts[3]:
140 parts[3] = parts[3].split(':', 1)[0]
141 packagekit_id = ';'.join(parts)
142 versions[packagekit_id] = info
143 tran = _PackageKitTransaction(self.pk, details_cb, error_cb)
144 tran.proxy.GetDetails(versions.keys())
145 else:
146 _logger_pk.info(_('Empty resolve for %s'), package_names)
147 blocker.trigger()
149 # Send queries
150 blocker = tasks.Blocker('PackageKit %s' % package_names)
151 for package in package_names:
152 self._candidates[package] = blocker
154 try:
155 _logger_pk.debug(_('Ask for %s'), package_names)
156 tran = _PackageKitTransaction(self.pk, resolve_cb, error_cb)
157 tran.proxy.Resolve('none', package_names)
159 in_progress.append(blocker)
160 except:
161 __, ex, tb = sys.exc_info()
162 blocker.trigger((ex, tb))
163 raise
165 # Now we've collected all the requests together, split them up into chunks
166 # that PackageKit can handle ( < 100 per batch )
167 #_logger_pk.info("sending %d packages", len(package_names))
168 while package_names:
169 next_batch = package_names[:MAX_PACKAGE_KIT_TRANSACTION_SIZE]
170 package_names = package_names[MAX_PACKAGE_KIT_TRANSACTION_SIZE:]
171 do_batch(next_batch)
173 while in_progress:
174 try:
175 yield in_progress
176 except:
177 pass
178 in_progress = [b for b in in_progress if not b.happened]
180 class PackageKitDownload:
181 def __init__(self, url, hint, pk, packagekit_id, expected_size):
182 self.url = url
183 self.status = download.download_fetching
184 self.hint = hint
185 self.aborted_by_user = False
187 self.downloaded = None
189 self.expected_size = expected_size
191 self.packagekit_id = packagekit_id
192 self._impl = hint
193 self._transaction = None
194 self.pk = pk
196 def error_cb(sender):
197 self.status = download.download_failed
198 ex = SafeException('PackageKit install failed: %s' % (sender.error_details or sender.error_code))
199 self.downloaded.trigger(exception = (ex, None))
201 def installed_cb(sender):
202 self._impl.installed = True;
203 self.status = download.download_complete
204 self.downloaded.trigger()
206 def install_packages():
207 package_name = self.packagekit_id
208 self._transaction = _PackageKitTransaction(self.pk, installed_cb, error_cb)
209 self._transaction.compat_call([
210 ('InstallPackages', [package_name]),
211 ('InstallPackages', False, [package_name]),
214 _auth_wrapper(install_packages)
216 self.downloaded = tasks.Blocker('PackageKit install %s' % self.packagekit_id)
218 def abort(self):
219 _logger_pk.debug(_('Cancel transaction'))
220 self.aborted_by_user = True
221 self._transaction.proxy.Cancel()
222 self.status = download.download_failed
223 self.downloaded.trigger()
225 def get_current_fraction(self):
226 if self._transaction is None:
227 return None
228 percentage = self._transaction.getPercentage()
229 if percentage > 100:
230 return None
231 else:
232 return float(percentage) / 100.
234 def get_bytes_downloaded_so_far(self):
235 fraction = self.get_current_fraction()
236 if fraction is None:
237 return 0
238 else:
239 if self.expected_size is None:
240 return 0
241 return int(self.expected_size * fraction)
243 def _auth_wrapper(method, *args):
244 try:
245 return method(*args)
246 except dbus.exceptions.DBusException as e:
247 if e.get_dbus_name() != \
248 'org.freedesktop.PackageKit.Transaction.RefusedByPolicy':
249 raise
251 iface, auth = e.get_dbus_message().split()
252 if not auth.startswith('auth_'):
253 raise
255 _logger_pk.debug(_('Authentication required for %s'), auth)
257 pk_auth = dbus.SessionBus().get_object(
258 'org.freedesktop.PolicyKit.AuthenticationAgent', '/',
259 'org.gnome.PolicyKit.AuthorizationManager.SingleInstance')
261 if not pk_auth.ObtainAuthorization(iface, dbus.UInt32(0),
262 dbus.UInt32(os.getpid()), timeout=300):
263 raise
265 return method(*args)
267 class _PackageKitTransaction(object):
268 def __init__(self, pk, finished_cb=None, error_cb=None):
269 self._finished_cb = finished_cb
270 self._error_cb = error_cb
271 self.error_code = None
272 self.error_details = None
273 self.package = {}
274 self.details = {}
275 self.files = {}
277 self.object = dbus.SystemBus().get_object(
278 'org.freedesktop.PackageKit', pk.GetTid(), False)
279 self.proxy = dbus.Interface(self.object,
280 'org.freedesktop.PackageKit.Transaction')
281 self._props = dbus.Interface(self.object, dbus.PROPERTIES_IFACE)
283 self._signals = []
284 for signal, cb in [('Finished', self.__finished_cb),
285 ('ErrorCode', self.__error_code_cb),
286 ('StatusChanged', self.__status_changed_cb),
287 ('Package', self.__package_cb),
288 ('Details', self.__details_cb),
289 ('Files', self.__files_cb)]:
290 self._signals.append(self.proxy.connect_to_signal(signal, cb))
292 defaultlocale = locale.getdefaultlocale()[0]
293 if defaultlocale is not None:
294 self.compat_call([
295 ('SetLocale', defaultlocale),
296 ('SetHints', ['locale=%s' % defaultlocale]),
299 def getPercentage(self):
300 result = self.get_prop('Percentage')
301 if result is None:
302 result, __, __, __ = self.proxy.GetProgress()
303 return result
305 def get_prop(self, prop, default = None):
306 try:
307 return self._props.Get('org.freedesktop.PackageKit.Transaction', prop)
308 except:
309 return default
311 def compat_call(self, calls):
312 for call in calls:
313 method = call[0]
314 args = call[1:]
315 try:
316 dbus_method = self.proxy.get_dbus_method(method)
317 return dbus_method(*args)
318 except dbus.exceptions.DBusException as e:
319 if e.get_dbus_name() != \
320 'org.freedesktop.DBus.Error.UnknownMethod':
321 raise
322 raise Exception('Cannot call %r DBus method' % calls)
324 def __finished_cb(self, exit, runtime):
325 _logger_pk.debug(_('Transaction finished: %s'), exit)
327 for i in self._signals:
328 i.remove()
330 if self.error_code is not None:
331 self._error_cb(self)
332 else:
333 self._finished_cb(self)
335 def __error_code_cb(self, code, details):
336 _logger_pk.info(_('Transaction failed: %s(%s)'), details, code)
337 self.error_code = code
338 self.error_details = details
340 def __package_cb(self, status, id, summary):
341 from zeroinstall.injector import distro
343 package_name, version, arch, repo_ = id.split(';')
344 clean_version = distro.try_cleanup_distro_version(version)
345 if not clean_version:
346 _logger_pk.warn(_("Can't parse distribution version '%(version)s' for package '%(package)s'"), {'version': version, 'package': package_name})
347 clean_arch = distro.canonical_machine(arch)
348 package = {'version': clean_version,
349 'name': package_name,
350 'arch': clean_arch,
351 'installed': (status == 'installed')}
352 _logger_pk.debug(_('Package: %s %r'), id, package)
353 self.package[str(id)] = package
355 def __details_cb(self, id, licence, group, detail, url, size):
356 details = {'licence': str(licence),
357 'group': str(group),
358 'detail': str(detail),
359 'url': str(url),
360 'size': int(size)}
361 _logger_pk.debug(_('Details: %s %r'), id, details)
362 self.details[id] = details
364 def __files_cb(self, id, files):
365 self.files[id] = files.split(';')
367 def __status_changed_cb(self, status):
368 pass