Commit 7a4e684a authored by segfault's avatar segfault

Remove unused module packagekit.py

parent 75f75992
# Based on isenkram, copyright (C) 2013,2015-2016 Petter Reinholdtsen <pere@debian.org>
import logging
from pydbus import SystemBus
import gi
from gi.repository import GLib
gi.require_version('PackageKitGlib', '1.0')
from gi.repository import PackageKitGlib as packagekit
from typing import List
logger = logging.getLogger(__name__)
class PackageException(Exception):
"""A package operation has failed."""
def __init__(self, error_string=None, error_details=None, *args):
"""Store packagekit error string and details."""
super().__init__(*args)
self.error_string = error_string
self.error_details = error_details
def __str__(self):
"""Return the strin representation of the exception."""
return 'PackageException(error_string="{0}", error_details="{1}")'.format(self.error_string, self.error_details)
class PackageKitManager(object):
"""Helper to install and uninstall packages using PackageKit.
Reference for PackageKit DBus API: https://www.freedesktop.org/software/PackageKit/gtk-doc/api-reference.html"""
def __init__(self, status_callback: callable, percentage_callback: callable):
self.status_callback = status_callback
self.percentage_callback = percentage_callback
self.bus = SystemBus()
self.packagekit = self.bus.get("org.freedesktop.PackageKit")
self.client = packagekit.Client()
self.client.set_interactive(False)
# Progress
self.role = None
self.status = None
self.package = None
self.percentage = None
def install(self, package_names: List[str]):
"""Run a PackageKit transaction to install given packages."""
try:
self.do_install(package_names)
except GLib.Error as exception:
raise PackageException(exception.message)
def do_install(self, package_names: List[str]):
"""Run a PackageKit transaction to install given packages.
Raise exception in case of error.
"""
# Refresh package cache from all enabled repositories
results = self.client.refresh_cache(
False, # don't enforce reloading if there is valid, up to date data
None, # cancellable
self.progress_callback, # progress callback
None # progress callback user data
)
self.assert_success(results)
package_ids = self.get_package_ids(package_names, package_filter=0)
# Start package installation
results = self.client.install_packages(
1 << packagekit.TransactionFlagEnum.ONLY_TRUSTED, # only install signed packages
package_ids, # packages to install
None, # cancellable
self.progress_callback, # progress callback
None # progress callback user data
)
self.assert_success(results)
def uninstall(self, package_names: List[str]):
"""Run a PackageKit transaction to uninstall given packages."""
try:
self.do_uninstall(package_names)
except GLib.Error as exception:
raise PackageException(exception.message)
def do_uninstall(self, package_names: List[str]):
"""Run a PackageKit transaction to uninstall given packages.
Raise exception in case of error.
XXX: This does not remove config files of the package (as `apt remove --purge` would do),
and it seems like there is currently no way to achieve this with packagekit.
Is removing the config files important to us?
"""
package_ids = self.get_package_ids(package_names, package_filter=(1 << packagekit.FilterEnum.INSTALLED))
# Start package uninstallation
results = self.client.remove_packages(
0, # transaction flags
package_ids, # packages to uninstall
True, # allow removing dependent packages
True, # remove dependencies installed along with this pkg
None, # cancellable
self.progress_callback, # progress callback
None # progress callback user data
)
self.assert_success(results)
def get_package_ids(self, package_names: List[str], package_filter: int) -> List[object]:
results = self.client.resolve(
package_filter, # package filter flags (for example to resolve only installed packages)
tuple(package_names), # names of packages to resolve
None, # cancellable
self.progress_callback, # progress callback
None # progress callback user data
)
self.assert_success(results)
packages_resolved = {package.get_name(): package for package in results.get_package_array()}
for package_name in package_names:
if package_name not in packages_resolved or not packages_resolved[package_name]:
raise PackageException("Package %r not found" % package_name)
return [packages_resolved[package_name].get_id() for package_name in package_names]
@staticmethod
def assert_success(results):
"""Check that the most recent operation was a success."""
if results and results.get_error_code() is not None:
error = results.get_error_code()
error_code = error.get_code() if error else None
error_string = packagekit.ErrorEnum.to_string(error_code) if error_code else None
error_details = error.get_details() if error else None
raise PackageException(error_string, error_details)
def progress_callback(self, progress, progress_type, user_data=None):
"""Process progress updates on package resolve operation"""
if progress_type == packagekit.ProgressType.ROLE:
self.role = packagekit.RoleEnum.to_string(progress.props.role)
logger.debug("Progress update: Role = %r", self.role)
self.status = str()
self.package = str()
self.percentage = -1
self.status_callback(self.role)
elif progress_type == packagekit.ProgressType.STATUS:
self.status = packagekit.StatusEnum.to_string(progress.props.status)
logger.debug("Progress update: Status = %r", self.status)
self.status_callback(self.role + " - " + self.status)
elif progress_type == packagekit.ProgressType.PACKAGE:
self.package = progress.props.package.get_name()
logger.debug("Progress update: Package = %r", self.package)
self.status_callback(self.role + " - " + self.status + " - " + self.package)
elif progress_type == packagekit.ProgressType.PERCENTAGE:
self.percentage = progress.props.percentage
logger.debug("Progress update: Percentage = %r", self.percentage)
self.percentage_callback(self.percentage)
return
else:
logger.debug("Progress update: Unhandled progress type %r", progress_type)
return
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment