summaryrefslogtreecommitdiff
path: root/apt
diff options
context:
space:
mode:
authorMichael Vogt <michael.vogt@ubuntu.com>2010-02-17 16:17:36 +0100
committerMichael Vogt <michael.vogt@ubuntu.com>2010-02-17 16:17:36 +0100
commitd1b2c0fbd34cee0ffced8270149b87def7675b2a (patch)
treef4b8a14fc619b2e6ea297385e289b2549f6f21a4 /apt
parentc62bdd0b1b633b134ba551847da0c1e8d12115c2 (diff)
parent930f6a2899b0b410777397fb206a8eba8c99100c (diff)
downloadpython-apt-d1b2c0fbd34cee0ffced8270149b87def7675b2a.tar.gz
merged from the debian-sid branch
Diffstat (limited to 'apt')
-rw-r--r--apt/__init__.py17
-rw-r--r--apt/cache.py425
-rw-r--r--apt/cdrom.py32
-rw-r--r--apt/debfile.py118
-rw-r--r--apt/deprecation.py106
-rw-r--r--apt/package.py740
-rw-r--r--apt/progress/__init__.py420
-rw-r--r--apt/progress/base.py302
-rw-r--r--apt/progress/gtk2.py239
-rw-r--r--apt/progress/old.py205
-rw-r--r--apt/progress/text.py261
-rw-r--r--apt/utils.py38
12 files changed, 1853 insertions, 1050 deletions
diff --git a/apt/__init__.py b/apt/__init__.py
index 4b3e6bf2..677a625b 100644
--- a/apt/__init__.py
+++ b/apt/__init__.py
@@ -17,22 +17,21 @@
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA
# import the core of apt_pkg
+"""High-Level Interface for working with apt."""
import apt_pkg
-import sys
-import os
# import some fancy classes
from apt.package import Package
from apt.cache import Cache, ProblemResolver
-from apt.progress import (
- OpProgress, FetchProgress, InstallProgress, CdromProgress)
from apt.cdrom import Cdrom
-from apt_pkg import SizeToStr, TimeToStr, VersionCompare
+
+if apt_pkg._COMPAT_0_7:
+ from apt.progress.old import (OpProgress, FetchProgress, InstallProgress,
+ CdromProgress)
+ from apt_pkg import (size_to_str as SizeToStr, time_to_str as TimeToStr,
+ version_compare as VersionCompare)
# init the package system
apt_pkg.init()
-
-#import warnings
-#warnings.warn("apt API not stable yet", FutureWarning)
-#del warnings
+__all__ = ['Cache', 'Cdrom', 'Package']
diff --git a/apt/cache.py b/apt/cache.py
index ae6f2fcc..24d63361 100644
--- a/apt/cache.py
+++ b/apt/cache.py
@@ -24,7 +24,10 @@ import weakref
import apt_pkg
from apt import Package
-import apt.progress
+from apt_pkg import gettext as _
+from apt.deprecation import (AttributeDeprecatedBy, function_deprecated_by,
+ deprecated_args)
+import apt.progress.text
class FetchCancelledException(IOError):
@@ -59,23 +62,23 @@ class Cache(object):
self._set = set()
if memonly:
# force apt to build its caches in memory
- apt_pkg.Config.Set("Dir::Cache::pkgcache", "")
+ apt_pkg.config.set("Dir::Cache::pkgcache", "")
if rootdir:
if os.path.exists(rootdir+"/etc/apt/apt.conf"):
- apt_pkg.ReadConfigFile(apt_pkg.Config,
+ apt_pkg.read_config_file(apt_pkg.config,
rootdir + "/etc/apt/apt.conf")
if os.path.isdir(rootdir+"/etc/apt/apt.conf.d"):
- apt_pkg.ReadConfigDir(apt_pkg.Config,
+ apt_pkg.read_config_dir(apt_pkg.config,
rootdir + "/etc/apt/apt.conf.d")
- apt_pkg.Config.Set("Dir", rootdir)
- apt_pkg.Config.Set("Dir::State::status",
+ apt_pkg.config.set("Dir", rootdir)
+ apt_pkg.config.set("Dir::State::status",
rootdir + "/var/lib/dpkg/status")
# create required dirs/files when run with special rootdir
# automatically
self._check_and_create_required_dirs(rootdir)
# Call InitSystem so the change to Dir::State::Status is actually
# recognized (LP: #320665)
- apt_pkg.InitSystem()
+ apt_pkg.init_system()
self.open(progress)
def _check_and_create_required_dirs(self, rootdir):
@@ -99,7 +102,7 @@ class Cache(object):
if not os.path.exists(rootdir+f):
open(rootdir+f,"w")
- def _runCallbacks(self, name):
+ def _run_callbacks(self, name):
""" internal helper to run a callback """
if name in self._callbacks:
for callback in self._callbacks[name]:
@@ -110,31 +113,32 @@ class Cache(object):
a dictionary
"""
if progress is None:
- progress = apt.progress.OpProgress()
- self._runCallbacks("cache_pre_open")
- self._cache = apt_pkg.GetCache(progress)
- self._depcache = apt_pkg.GetDepCache(self._cache)
- self._records = apt_pkg.GetPkgRecords(self._cache)
- self._list = apt_pkg.GetPkgSourceList()
- self._list.ReadMainList()
+ progress = apt.progress.text.OpProgress()
+ self._run_callbacks("cache_pre_open")
+
+ self._cache = apt_pkg.Cache(progress)
+ self._depcache = apt_pkg.DepCache(self._cache)
+ self._records = apt_pkg.PackageRecords(self._cache)
+ self._list = apt_pkg.SourceList()
+ self._list.read_main_list()
self._set.clear()
self._weakref.clear()
- progress.Op = "Building data structures"
+ progress.op = _("Building data structures")
i=last=0
- size=len(self._cache.Packages)
- for pkg in self._cache.Packages:
+ size=len(self._cache.packages)
+ for pkg in self._cache.packages:
if progress is not None and last+100 < i:
progress.update(i/float(size)*100)
last=i
# drop stuff with no versions (cruft)
- if len(pkg.VersionList) > 0:
- self._set.add(pkg.Name)
+ if len(pkg.version_list) > 0:
+ self._set.add(pkg.name)
i += 1
progress.done()
- self._runCallbacks("cache_post_open")
+ self._run_callbacks("cache_post_open")
def __getitem__(self, key):
""" look like a dictionary (get key) """
@@ -165,97 +169,105 @@ class Cache(object):
def keys(self):
return list(self._set)
- def getChanges(self):
+ def get_changes(self):
""" Get the marked changes """
changes = []
- for p in self:
- if p.markedUpgrade or p.markedInstall or p.markedDelete or \
- p.markedDowngrade or p.markedReinstall:
- changes.append(p)
+ for pkg in self:
+ if (pkg.marked_upgrade or pkg.marked_install or pkg.marked_delete
+ or pkg.marked_downgrade or pkg.marked_reinstall):
+ changes.append(pkg)
return changes
- def upgrade(self, distUpgrade=False):
- """ Upgrade the all package, DistUpgrade will also install
- new dependencies
+ @deprecated_args
+ def upgrade(self, dist_upgrade=False):
+ """Upgrade all packages.
+
+ If the parameter *dist_upgrade* is True, new dependencies will be
+ installed as well (and conflicting packages may be removed). The
+ default value is False.
"""
- self.cachePreChange()
- self._depcache.Upgrade(distUpgrade)
- self.cachePostChange()
+ self.cache_pre_change()
+ self._depcache.upgrade(dist_upgrade)
+ self.cache_post_change()
@property
- def requiredDownload(self):
+ def required_download(self):
"""Get the size of the packages that are required to download."""
- pm = apt_pkg.GetPackageManager(self._depcache)
- fetcher = apt_pkg.GetAcquire()
- pm.GetArchives(fetcher, self._list, self._records)
- return fetcher.FetchNeeded
+ pm = apt_pkg.PackageManager(self._depcache)
+ fetcher = apt_pkg.Acquire()
+ pm.get_archives(fetcher, self._list, self._records)
+ return fetcher.fetch_needed
@property
- def additionalRequiredSpace(self):
+ def required_space(self):
"""Get the size of the additional required space on the fs."""
- return self._depcache.UsrSize
+ return self._depcache.usr_size
@property
- def reqReinstallPkgs(self):
+ def req_reinstall_pkgs(self):
"""Return the packages not downloadable packages in reqreinst state."""
reqreinst = set()
for pkg in self:
if (not pkg.candidate.downloadable and
- (pkg._pkg.InstState == apt_pkg.InstStateReInstReq or
- pkg._pkg.InstState == apt_pkg.InstStateHoldReInstReq)):
+ (pkg._pkg.inst_state == apt_pkg.INSTSTATE_REINSTREQ or
+ pkg._pkg.inst_state == apt_pkg.INSTSTATE_HOLD_REINSTREQ)):
reqreinst.add(pkg.name)
return reqreinst
- def _runFetcher(self, fetcher):
+ def _run_fetcher(self, fetcher):
# do the actual fetching
- res = fetcher.Run()
+ res = fetcher.run()
# now check the result (this is the code from apt-get.cc)
failed = False
transient = False
- errMsg = ""
- for item in fetcher.Items:
- if item.Status == item.StatDone:
+ err_msg = ""
+ for item in fetcher.items:
+ if item.status == item.STAT_DONE:
continue
- if item.StatIdle:
+ if item.STAT_IDLE:
transient = True
continue
- errMsg += "Failed to fetch %s %s\n" % (item.DescURI,
- item.ErrorText)
+ err_msg += "Failed to fetch %s %s\n" % (item.desc_uri,
+ item.error_text)
failed = True
# we raise a exception if the download failed or it was cancelt
- if res == fetcher.ResultCancelled:
- raise FetchCancelledException(errMsg)
+ if res == fetcher.RESULT_CANCELLED:
+ raise FetchCancelledException(err_msg)
elif failed:
- raise FetchFailedException(errMsg)
+ raise FetchFailedException(err_msg)
return res
- def _fetchArchives(self, fetcher, pm):
+ def _fetch_archives(self, fetcher, pm):
""" fetch the needed archives """
# get lock
- lockfile = apt_pkg.Config.FindDir("Dir::Cache::Archives") + "lock"
- lock = apt_pkg.GetLock(lockfile)
+ lockfile = apt_pkg.config.find_dir("Dir::Cache::Archives") + "lock"
+ lock = apt_pkg.get_lock(lockfile)
if lock < 0:
raise LockFailedException("Failed to lock %s" % lockfile)
try:
# this may as well throw a SystemError exception
- if not pm.GetArchives(fetcher, self._list, self._records):
+ if not pm.get_archives(fetcher, self._list, self._records):
return False
# now run the fetcher, throw exception if something fails to be
# fetched
- return self._runFetcher(fetcher)
+ return self._run_fetcher(fetcher)
finally:
os.close(lock)
- def isVirtualPackage(self, pkgname):
+ def is_virtual_package(self, pkgname):
"""Return whether the package is a virtual package."""
- pkg = self._cache[pkgname]
- return bool(pkg.ProvidesList and not pkg.VersionList)
+ try:
+ pkg = self._cache[pkgname]
+ except KeyError:
+ return False
+ else:
+ return bool(pkg.provides_list and not pkg.version_list)
- def getProvidingPackages(self, virtual):
+ def get_providing_packages(self, virtual):
"""
Return a list of packages which provide the virtual package of the
specified name
@@ -263,49 +275,81 @@ class Cache(object):
providers = []
try:
vp = self._cache[virtual]
- if len(vp.VersionList) != 0:
+ if len(vp.version_list) != 0:
return providers
except KeyError:
return providers
for pkg in self:
- v = self._depcache.GetCandidateVer(pkg._pkg)
+ v = self._depcache.get_candidate_ver(pkg._pkg)
if v is None:
continue
- for p in v.ProvidesList:
+ for p in v.provides_list:
if virtual == p[0]:
# we found a pkg that provides this virtual pkg
providers.append(pkg)
return providers
- def update(self, fetchProgress=None, pulseInterval=0, raiseOnError=False):
- " run the equivalent of apt-get update "
- lockfile = apt_pkg.Config.FindDir("Dir::State::Lists") + "lock"
- lock = apt_pkg.GetLock(lockfile)
+ @deprecated_args
+ def update(self, fetch_progress=None, pulse_interval=0,
+ raise_on_error=True):
+ """Run the equivalent of apt-get update.
+
+ The first parameter *fetch_progress* may be set to an instance of
+ apt.progress.FetchProgress, the default is apt.progress.FetchProgress()
+ .
+ """
+ lockfile = apt_pkg.config.find_dir("Dir::State::Lists") + "lock"
+ lock = apt_pkg.get_lock(lockfile)
+
if lock < 0:
raise LockFailedException("Failed to lock %s" % lockfile)
try:
- if fetchProgress is None:
- fetchProgress = apt.progress.FetchProgress()
- res = self._cache.Update(fetchProgress, self._list, pulseInterval)
-
- if res == 2 and raiseOnError:
- raise FetchCancelledException()
- elif res == 1 and raiseOnError:
+ if fetch_progress is None:
+ fetch_progress = apt.progress.base.AcquireProgress()
+ try:
+ res = self._cache.update(fetch_progress, self._list,
+ pulse_interval)
+ except SystemError, e:
+ raise FetchFailedException(e)
+ if not res and raise_on_error:
raise FetchFailedException()
else:
return res
finally:
os.close(lock)
- def installArchives(self, pm, installProgress):
- installProgress.startUpdate()
- res = installProgress.run(pm)
- installProgress.finishUpdate()
+ @deprecated_args
+ def install_archives(self, pm, install_progress):
+ """
+ The first parameter *pm* refers to an object returned by
+ apt_pkg.PackageManager().
+
+ The second parameter *install_progress* refers to an InstallProgress()
+ object of the module apt.progress.
+ """
+ try:
+ install_progress.start_update()
+ except AttributeError:
+ install_progress.startUpdate()
+ res = install_progress.run(pm)
+ try:
+ install_progress.finish_update()
+ except AttributeError:
+ install_progress.finishUpdate()
return res
- def commit(self, fetchProgress=None, installProgress=None):
- """ Apply the marked changes to the cache """
+ @deprecated_args
+ def commit(self, fetch_progress=None, install_progress=None):
+ """Apply the marked changes to the cache.
+
+ The first parameter, *fetch_progress*, refers to a FetchProgress()
+ object as found in apt.progress, the default being
+ apt.progress.FetchProgress().
+
+ The second parameter, *install_progress*, is a
+ apt.progress.InstallProgress() object.
+ """
# FIXME:
# use the new acquire/pkgmanager interface here,
# raise exceptions when a download or install fails
@@ -313,45 +357,52 @@ class Cache(object):
# Current a failed download will just display "error"
# which is less than optimal!
- if fetchProgress is None:
- fetchProgress = apt.progress.FetchProgress()
- if installProgress is None:
- installProgress = apt.progress.InstallProgress()
+ if fetch_progress is None:
+ fetch_progress = apt.progress.base.AcquireProgress()
+ if install_progress is None:
+ install_progress = apt.progress.base.InstallProgress()
- pm = apt_pkg.GetPackageManager(self._depcache)
- fetcher = apt_pkg.GetAcquire(fetchProgress)
+ pm = apt_pkg.PackageManager(self._depcache)
+ fetcher = apt_pkg.Acquire(fetch_progress)
while True:
# fetch archives first
- res = self._fetchArchives(fetcher, pm)
+ res = self._fetch_archives(fetcher, pm)
# then install
- res = self.installArchives(pm, installProgress)
- if res == pm.ResultCompleted:
+ res = self.install_archives(pm, install_progress)
+ if res == pm.RESULT_COMPLETED:
break
- elif res == pm.ResultFailed:
+ elif res == pm.RESULT_FAILED:
raise SystemError("installArchives() failed")
- elif res == pm.ResultIncomplete:
+ elif res == pm.RESULT_INCOMPLETE:
pass
else:
raise SystemError("internal-error: unknown result code from InstallArchives: %s" % res)
# reload the fetcher for media swaping
- fetcher.Shutdown()
- return (res == pm.ResultCompleted)
+ fetcher.shutdown()
+ return (res == pm.RESULT_COMPLETED)
def clear(self):
""" Unmark all changes """
- self._depcache.Init()
+ self._depcache.init()
# cache changes
- def cachePostChange(self):
+ def cache_post_change(self):
" called internally if the cache has changed, emit a signal then "
- self._runCallbacks("cache_post_change")
+ self._run_callbacks("cache_post_change")
- def cachePreChange(self):
+ def cache_pre_change(self):
""" called internally if the cache is about to change, emit
a signal then """
- self._runCallbacks("cache_pre_change")
+ self._run_callbacks("cache_pre_change")
+
+ def connect(self, name, callback):
+ """ connect to a signal, currently only used for
+ cache_{post,pre}_{changed,open} """
+ if not name in self._callbacks:
+ self._callbacks[name] = []
+ self._callbacks[name].append(callback)
def actiongroup(self):
"""Return an ActionGroup() object for the current cache.
@@ -359,35 +410,54 @@ class Cache(object):
Action groups can be used to speedup actions. The action group is
active as soon as it is created, and disabled when the object is
deleted or when release() is called.
- """
- return apt_pkg.GetPkgActionGroup(self._depcache)
- def connect(self, name, callback):
- """ connect to a signal, currently only used for
- cache_{post,pre}_{changed,open} """
- if not name in self._callbacks:
- self._callbacks[name] = []
- self._callbacks[name].append(callback)
+ You can use the action group as a context manager, this is the
+ recommended way::
+
+ with cache.actiongroup():
+ for package in my_selected_packages:
+ package.mark_install()
+
+ This way, the ActionGroup is automatically released as soon as the
+ with statement block is left. It also has the benefit of making it
+ clear which parts of the code run with a action group and which
+ don't.
+ """
+ return apt_pkg.ActionGroup(self._depcache)
@property
def broken_count(self):
"""Return the number of packages with broken dependencies."""
- return self._depcache.BrokenCount
+ return self._depcache.broken_count
@property
def delete_count(self):
"""Return the number of packages marked for deletion."""
- return self._depcache.DelCount
+ return self._depcache.del_count
@property
def install_count(self):
"""Return the number of packages marked for installation."""
- return self._depcache.InstCount
+ return self._depcache.inst_count
@property
def keep_count(self):
"""Return the number of packages marked as keep."""
- return self._depcache.KeepCount
+ return self._depcache.keep_count
+
+ if apt_pkg._COMPAT_0_7:
+ _runCallbacks = function_deprecated_by(_run_callbacks)
+ getChanges = function_deprecated_by(get_changes)
+ requiredDownload = AttributeDeprecatedBy('required_download')
+ additionalRequiredSpace = AttributeDeprecatedBy('required_space')
+ reqReinstallPkgs = AttributeDeprecatedBy('req_reinstall_pkgs')
+ _runFetcher = function_deprecated_by(_run_fetcher)
+ _fetchArchives = function_deprecated_by(_fetch_archives)
+ isVirtualPackage = function_deprecated_by(is_virtual_package)
+ getProvidingPackages = function_deprecated_by(get_providing_packages)
+ installArchives = function_deprecated_by(install_archives)
+ cachePostChange = function_deprecated_by(cache_post_change)
+ cachePreChange = function_deprecated_by(cache_pre_change)
class ProblemResolver(object):
@@ -397,31 +467,31 @@ class ProblemResolver(object):
"""
def __init__(self, cache):
- self._resolver = apt_pkg.GetPkgProblemResolver(cache._depcache)
+ self._resolver = apt_pkg.ProblemResolver(cache._depcache)
def clear(self, package):
"""Reset the package to the default state."""
- self._resolver.Clear(package._pkg)
+ self._resolver.clear(package._pkg)
def install_protect(self):
"""mark protected packages for install or removal."""
- self._resolver.InstallProtect()
+ self._resolver.install_protect()
def protect(self, package):
"""Protect a package so it won't be removed."""
- self._resolver.Protect(package._pkg)
+ self._resolver.protect(package._pkg)
def remove(self, package):
"""Mark a package for removal."""
- self._resolver.Remove(package._pkg)
+ self._resolver.remove(package._pkg)
def resolve(self):
"""Resolve dependencies, try to remove packages where needed."""
- self._resolver.Resolve()
+ self._resolver.resolve()
def resolve_by_keep(self):
"""Resolve dependencies, do not try to remove packages."""
- self._resolver.ResolveByKeep()
+ self._resolver.resolve_by_keep()
# ----------------------------- experimental interface
@@ -441,7 +511,7 @@ class MarkedChangesFilter(Filter):
""" Filter that returns all marked changes """
def apply(self, pkg):
- if pkg.markedInstall or pkg.markedDelete or pkg.markedUpgrade:
+ if pkg.marked_install or pkg.marked_delete or pkg.marked_upgrade:
return True
else:
return False
@@ -458,8 +528,8 @@ class FilteredCache(object):
self.cache = Cache(progress)
else:
self.cache = cache
- self.cache.connect("cache_post_change", self.filterCachePostChange)
- self.cache.connect("cache_post_open", self.filterCachePostChange)
+ self.cache.connect("cache_post_change", self.filter_cache_post_change)
+ self.cache.connect("cache_post_open", self.filter_cache_post_change)
self._filtered = {}
self._filters = []
@@ -482,7 +552,7 @@ class FilteredCache(object):
def __contains__(self, key):
return (key in self._filtered)
- def _reapplyFilter(self):
+ def _reapply_filter(self):
" internal helper to refilter "
self._filtered = {}
for pkg in self.cache:
@@ -491,18 +561,19 @@ class FilteredCache(object):
self._filtered[pkg.name] = 1
break
- def setFilter(self, filter):
+ def set_filter(self, filter):
"""Set the current active filter."""
self._filters = []
self._filters.append(filter)
#self._reapplyFilter()
# force a cache-change event that will result in a refiltering
- self.cache.cachePostChange()
+ self.cache.cache_post_change()
- def filterCachePostChange(self):
+ def filter_cache_post_change(self):
"""Called internally if the cache changes, emit a signal then."""
#print "filterCachePostChange()"
- self._reapplyFilter()
+ self._reapply_filter()
+
# def connect(self, name, callback):
# self.cache.connect(name, callback)
@@ -512,6 +583,12 @@ class FilteredCache(object):
#print "getattr: %s " % key
return getattr(self.cache, key)
+ if apt_pkg._COMPAT_0_7:
+ _reapplyFilter = function_deprecated_by(_reapply_filter)
+ setFilter = function_deprecated_by(set_filter)
+ filterCachePostChange = function_deprecated_by(\
+ filter_cache_post_change)
+
def cache_pre_changed():
print "cache pre changed"
@@ -521,61 +598,61 @@ def cache_post_changed():
print "cache post changed"
-# internal test code
-if __name__ == "__main__":
+def _test():
+ """Internal test code."""
print "Cache self test"
apt_pkg.init()
- c = Cache(apt.progress.OpTextProgress())
- c.connect("cache_pre_change", cache_pre_changed)
- c.connect("cache_post_change", cache_post_changed)
- print ("aptitude" in c)
- p = c["aptitude"]
- print p.name
- print len(c)
-
- for pkg in c.keys():
- x= c[pkg].name
-
- c.upgrade()
- changes = c.getChanges()
+ cache = Cache(apt.progress.text.OpProgress())
+ cache.connect("cache_pre_change", cache_pre_changed)
+ cache.connect("cache_post_change", cache_post_changed)
+ print ("aptitude" in cache)
+ pkg = cache["aptitude"]
+ print pkg.name
+ print len(cache)
+
+ for pkgname in cache.keys():
+ assert cache[pkgname].name == pkgname
+
+ cache.upgrade()
+ changes = cache.get_changes()
print len(changes)
- for p in changes:
- #print p.name
- x = p.name
+ for pkg in changes:
+ assert pkg.name
+
# see if fetching works
- for d in ["/tmp/pytest", "/tmp/pytest/partial"]:
- if not os.path.exists(d):
- os.mkdir(d)
- apt_pkg.Config.Set("Dir::Cache::Archives", "/tmp/pytest")
- pm = apt_pkg.GetPackageManager(c._depcache)
- fetcher = apt_pkg.GetAcquire(apt.progress.TextFetchProgress())
- c._fetchArchives(fetcher, pm)
+ for dir in ["/tmp/pytest", "/tmp/pytest/partial"]:
+ if not os.path.exists(dir):
+ os.mkdir(dir)
+ apt_pkg.config.set("Dir::Cache::Archives", "/tmp/pytest")
+ pm = apt_pkg.PackageManager(cache._depcache)
+ fetcher = apt_pkg.Acquire(apt.progress.text.AcquireProgress())
+ cache._fetch_archives(fetcher, pm)
#sys.exit(1)
print "Testing filtered cache (argument is old cache)"
- f = FilteredCache(c)
- f.cache.connect("cache_pre_change", cache_pre_changed)
- f.cache.connect("cache_post_change", cache_post_changed)
- f.cache.upgrade()
- f.setFilter(MarkedChangesFilter())
- print len(f)
- for pkg in f.keys():
- #print c[pkg].name
- x = f[pkg].name
-
- print len(f)
+ filtered = FilteredCache(cache)
+ filtered.cache.connect("cache_pre_change", cache_pre_changed)
+ filtered.cache.connect("cache_post_change", cache_post_changed)
+ filtered.cache.upgrade()
+ filtered.set_filter(MarkedChangesFilter())
+ print len(filtered)
+ for pkgname in filtered.keys():
+ assert pkgname == filtered[pkg].name
+
+ print len(filtered)
print "Testing filtered cache (no argument)"
- f = FilteredCache(progress=apt.progress.OpTextProgress())
- f.cache.connect("cache_pre_change", cache_pre_changed)
- f.cache.connect("cache_post_change", cache_post_changed)
- f.cache.upgrade()
- f.setFilter(MarkedChangesFilter())
- print len(f)
- for pkg in f.keys():
- #print c[pkg].name
- x = f[pkg].name
-
- print len(f)
+ filtered = FilteredCache(progress=apt.progress.base.OpProgress())
+ filtered.cache.connect("cache_pre_change", cache_pre_changed)
+ filtered.cache.connect("cache_post_change", cache_post_changed)
+ filtered.cache.upgrade()
+ filtered.set_filter(MarkedChangesFilter())
+ print len(filtered)
+ for pkgname in filtered.keys():
+ assert pkgname == filtered[pkgname].name
+
+ print len(filtered)
+if __name__ == '__main__':
+ _test()
diff --git a/apt/cdrom.py b/apt/cdrom.py
index b52762ad..01caa12f 100644
--- a/apt/cdrom.py
+++ b/apt/cdrom.py
@@ -23,10 +23,11 @@
import glob
import apt_pkg
-from apt.progress import CdromProgress
+from apt.deprecation import AttributeDeprecatedBy
+from apt.progress.base import CdromProgress
-class Cdrom(object):
+class Cdrom(apt_pkg.Cdrom):
"""Support for apt-cdrom like features.
This class has several optional parameters for initialisation, which may
@@ -44,43 +45,44 @@ class Cdrom(object):
"""
def __init__(self, progress=None, mountpoint=None, nomount=True):
- self._cdrom = apt_pkg.GetCdrom()
+ apt_pkg.Cdrom.__init__(self)
if progress is None:
self._progress = CdromProgress()
else:
self._progress = progress
# see if we have a alternative mountpoint
if mountpoint is not None:
- apt_pkg.Config.Set("Acquire::cdrom::mount", mountpoint)
+ apt_pkg.config.set("Acquire::cdrom::mount", mountpoint)
# do not mess with mount points by default
if nomount:
- apt_pkg.Config.Set("APT::CDROM::NoMount", "true")
+ apt_pkg.config.set("APT::CDROM::NoMount", "true")
else:
- apt_pkg.Config.Set("APT::CDROM::NoMount", "false")
+ apt_pkg.config.set("APT::CDROM::NoMount", "false")
- def add(self):
+ def add(self, progress=None):
"""Add cdrom to the sources.list."""
- return self._cdrom.Add(self._progress)
+ return apt_pkg.Cdrom.add(self, progress or self._progress)
- def ident(self):
+ def ident(self, progress=None):
"""Identify the cdrom."""
- (res, ident) = self._cdrom.Ident(self._progress)
- if res:
- return ident
+ return apt_pkg.Cdrom.ident(self, progress or self._progress)
@property
- def inSourcesList(self):
+ def in_sources_list(self):
"""Check if the cdrom is already in the current sources.list."""
cd_id = self.ident()
if cd_id is None:
# FIXME: throw exception instead
return False
# Get a list of files
- src = glob.glob(apt_pkg.Config.FindDir("Dir::Etc::sourceparts") + '*')
- src.append(apt_pkg.Config.FindFile("Dir::Etc::sourcelist"))
+ src = glob.glob(apt_pkg.config.find_dir("Dir::Etc::sourceparts") + '*')
+ src.append(apt_pkg.config.find_file("Dir::Etc::sourcelist"))
# Check each file
for fname in src:
for line in open(fname):
if not line.lstrip().startswith("#") and cd_id in line:
return True
return False
+
+ if apt_pkg._COMPAT_0_7:
+ inSourcesList = AttributeDeprecatedBy('in_sources_list')
diff --git a/apt/debfile.py b/apt/debfile.py
index 0e455508..e27917d5 100644
--- a/apt/debfile.py
+++ b/apt/debfile.py
@@ -17,12 +17,12 @@
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA
"""Classes for working with locally available Debian packages."""
-from gettext import gettext as _
import os
import sys
import apt_inst
import apt_pkg
+from apt_pkg import gettext as _
# Constants for comparing the local package file with the version in the cache
@@ -52,12 +52,9 @@ class DebPackage(object):
def open(self, filename):
" open given debfile "
self.filename = filename
- if not apt_inst.arCheckMember(open(self.filename), "debian-binary"):
- raise NoDebArchiveException(_("This is not a valid DEB archive, "
- "missing '%s' member" %
- "debian-binary"))
- control = apt_inst.debExtractControl(open(self.filename))
- self._sections = apt_pkg.ParseSection(control)
+ self._debfile = apt_inst.DebFile(self.filename)
+ control = self._debfile.control.extractdata("control")
+ self._sections = apt_pkg.TagSection(control)
self.pkgname = self._sections["Package"]
def __getitem__(self, key):
@@ -67,19 +64,11 @@ class DebPackage(object):
def filelist(self):
"""return the list of files in the deb."""
files = []
-
- def extract_cb(what, name, *_):
- files.append(name)
-
- for member in self._supported_data_members:
- if apt_inst.arCheckMember(open(self.filename), member):
- try:
- apt_inst.debExtract(open(self.filename), extract_cb,
- member)
- break
- except SystemError:
- return [_("List of files for '%s' could not be read" %
- self.filename)]
+ try:
+ self._debfile.data.go(lambda item, data: files.append(item.name))
+ except SystemError:
+ return [_("List of files for '%s' could not be read" %
+ self.filename)]
return files
def _is_or_group_satisfied(self, or_group):
@@ -97,16 +86,16 @@ class DebPackage(object):
# check for virtual pkgs
if not depname in self._cache:
- if self._cache.isVirtualPackage(depname):
+ if self._cache.is_virtual_package(depname):
self._dbg(3, "_isOrGroupSatisfied(): %s is virtual dep" %
depname)
- for pkg in self._cache.getProvidingPackages(depname):
- if pkg.isInstalled:
+ for pkg in self._cache.get_providing_packages(depname):
+ if pkg.is_installed:
return True
continue
inst = self._cache[depname].installed
- if inst is not None and apt_pkg.CheckDep(inst.version, oper, ver):
+ if inst is not None and apt_pkg.check_dep(inst.version, oper, ver):
return True
return False
@@ -117,9 +106,9 @@ class DebPackage(object):
# if we don't have it in the cache, it may be virtual
if not depname in self._cache:
- if not self._cache.isVirtualPackage(depname):
+ if not self._cache.is_virtual_package(depname):
continue
- providers = self._cache.getProvidingPackages(depname)
+ providers = self._cache.get_providing_packages(depname)
# if a package just has a single virtual provider, we
# just pick that (just like apt)
if len(providers) != 1:
@@ -129,10 +118,10 @@ class DebPackage(object):
# now check if we can satisfy the deps with the candidate(s)
# in the cache
pkg = self._cache[depname]
- cand = self._cache._depcache.GetCandidateVer(pkg._pkg)
+ cand = self._cache._depcache.get_candidate_ver(pkg._pkg)
if not cand:
continue
- if not apt_pkg.CheckDep(cand.VerStr, oper, ver):
+ if not apt_pkg.check_dep(cand.ver_str, oper, ver):
continue
# check if we need to install it
@@ -158,9 +147,9 @@ class DebPackage(object):
(pkgname, ver, oper))
pkg = self._cache[pkgname]
- if pkg.isInstalled:
+ if pkg.is_installed:
pkgver = pkg.installed.version
- elif pkg.markedInstall:
+ elif pkg.marked_install:
pkgver = pkg.candidate.version
else:
return False
@@ -168,7 +157,7 @@ class DebPackage(object):
#print "ver: %s" % ver
#print "pkgver: %s " % pkgver
#print "oper: %s " % oper
- if (apt_pkg.CheckDep(pkgver, oper, ver) and not
+ if (apt_pkg.check_dep(pkgver, oper, ver) and not
self.replaces_real_pkg(pkgname, oper, ver)):
self._failure_string += _("Conflicts with the installed package "
"'%s'" % pkg.name)
@@ -191,8 +180,8 @@ class DebPackage(object):
if not depname in self._cache:
# FIXME: we have to check for virtual replaces here as
# well (to pass tests/gdebi-test8.deb)
- if self._cache.isVirtualPackage(depname):
- for pkg in self._cache.getProvidingPackages(depname):
+ if self._cache.is_virtual_package(depname):
+ for pkg in self._cache.get_providing_packages(depname):
self._dbg(3, "conflicts virtual check: %s" % pkg.name)
# P/C/R on virtal pkg, e.g. ftpd
if self.pkgname == pkg.name:
@@ -211,7 +200,7 @@ class DebPackage(object):
"""List of package names conflicting with this package."""
key = "Conflicts"
try:
- return apt_pkg.ParseDepends(self._sections[key])
+ return apt_pkg.parse_depends(self._sections[key])
except KeyError:
return []
@@ -222,7 +211,7 @@ class DebPackage(object):
# find depends
for key in "Depends", "PreDepends":
try:
- depends.extend(apt_pkg.ParseDepends(self._sections[key]))
+ depends.extend(apt_pkg.parse_depends(self._sections[key]))
except KeyError:
pass
return depends
@@ -232,7 +221,7 @@ class DebPackage(object):
"""List of virtual packages which are provided by this package."""
key = "Provides"
try:
- return apt_pkg.ParseDepends(self._sections[key])
+ return apt_pkg.parse_depends(self._sections[key])
except KeyError:
return []
@@ -241,7 +230,7 @@ class DebPackage(object):
"""List of packages which are replaced by this package."""
key = "Replaces"
try:
- return apt_pkg.ParseDepends(self._sections[key])
+ return apt_pkg.parse_depends(self._sections[key])
except KeyError:
return []
@@ -253,15 +242,15 @@ class DebPackage(object):
"""
self._dbg(3, "replacesPkg() %s %s %s" % (pkgname, oper, ver))
pkg = self._cache[pkgname]
- if pkg.isInstalled:
+ if pkg.is_installed:
pkgver = pkg.installed.version
- elif pkg.markedInstall:
+ elif pkg.marked_install:
pkgver = pkg.candidate.version
else:
pkgver = None
for or_group in self.replaces:
for (name, ver, oper) in or_group:
- if (name == pkgname and apt_pkg.CheckDep(pkgver, oper, ver)):
+ if (name == pkgname and apt_pkg.check_dep(pkgver, oper, ver)):
self._dbg(3, "we have a replaces in our package for the "
"conflict against '%s'" % (pkgname))
return True
@@ -298,7 +287,7 @@ class DebPackage(object):
else:
cachever = self._cache[pkgname].candidate.version
if cachever is not None:
- cmp = apt_pkg.VersionCompare(cachever, debver)
+ cmp = apt_pkg.version_compare(cachever, debver)
self._dbg(1, "CompareVersion(debver,instver): %s" % cmp)
if cmp == 0:
return VERSION_SAME
@@ -310,11 +299,11 @@ class DebPackage(object):
def check(self):
"""Check if the package is installable."""
- self._dbg(3, "checkDepends")
+ self._dbg(3, "check_depends")
# check arch
arch = self._sections["Architecture"]
- if arch != "all" and arch != apt_pkg.Config.Find("APT::Architecture"):
+ if arch != "all" and arch != apt_pkg.config.find("APT::Architecture"):
self._dbg(1, "ERROR: Wrong architecture dude!")
self._failure_string = _("Wrong architecture '%s'" % arch)
return False
@@ -341,7 +330,7 @@ class DebPackage(object):
if not self.check_conflicts():
return False
- if self._cache._depcache.BrokenCount > 0:
+ if self._cache._depcache.broken_count > 0:
self._failure_string = _("Failed to satisfy all dependencies "
"(broken cache)")
# clean the cache again
@@ -351,13 +340,13 @@ class DebPackage(object):
def satisfy_depends_str(self, dependsstr):
"""Satisfy the dependencies in the given string."""
- return self._satisfy_depends(apt_pkg.ParseDepends(dependsstr))
+ return self._satisfy_depends(apt_pkg.parse_depends(dependsstr))
def _satisfy_depends(self, depends):
"""Satisfy the dependencies."""
# turn off MarkAndSweep via a action group (if available)
try:
- _actiongroup = apt_pkg.GetPkgActionGroup(self._cache._depcache)
+ _actiongroup = apt_pkg.ActionGroup(self._cache._depcache)
except AttributeError:
pass
# check depends
@@ -371,7 +360,7 @@ class DebPackage(object):
# now try it out in the cache
for pkg in self._need_pkgs:
try:
- self._cache[pkg].markInstall(fromUser=False)
+ self._cache[pkg].mark_install(fromUser=False)
except SystemError, e:
self._failure_string = _("Cannot install '%s'" % pkg)
self._cache.clear()
@@ -396,7 +385,7 @@ class DebPackage(object):
remove = []
unauthenticated = []
for pkg in self._cache:
- if pkg.markedInstall or pkg.markedUpgrade:
+ if pkg.marked_install or pkg.marked_upgrade:
install.append(pkg.name)
# check authentication, one authenticated origin is enough
# libapt will skip non-authenticated origins then
@@ -405,7 +394,7 @@ class DebPackage(object):
authenticated |= origin.trusted
if not authenticated:
unauthenticated.append(pkg.name)
- if pkg.markedDelete:
+ if pkg.marked_delete:
remove.append(pkg.name)
return (install, remove, unauthenticated)
@@ -417,7 +406,7 @@ class DebPackage(object):
def install(self, install_progress=None):
"""Install the package."""
if install_progress is None:
- return os.system("dpkg -i %s" % self.filename)
+ return os.spawnlp(os.P_WAIT, "dpkg", "dpkg", "-i", self.filename)
else:
try:
install_progress.start_update()
@@ -458,23 +447,22 @@ class DscSrcPackage(DebPackage):
conflicts_tags = ["Build-Conflicts", "Build-Conflicts-Indep"]
fobj = open(file)
- tagfile = apt_pkg.ParseTagFile(fobj)
- sec = tagfile.Section
+ tagfile = apt_pkg.TagFile(fobj)
try:
- while tagfile.Step() == 1:
+ for sec in tagfile:
for tag in depends_tags:
- if not sec.has_key(tag):
+ if not tag in sec:
continue
- self._depends.extend(apt_pkg.ParseSrcDepends(sec[tag]))
+ self._depends.extend(apt_pkg.parse_src_depends(sec[tag]))
for tag in conflicts_tags:
- if not sec.has_key(tag):
+ if not tag in sec:
continue
- self._conflicts.extend(apt_pkg.ParseSrcDepends(sec[tag]))
- if sec.has_key('Source'):
+ self._conflicts.extend(apt_pkg.parse_src_depends(sec[tag]))
+ if 'Source' in sec:
self.pkgname = sec['Source']
- if sec.has_key('Binary'):
+ if 'Binary' in sec:
self.binaries = sec['Binary'].split(', ')
- if sec.has_key('Version'):
+ if 'Version' in sec:
self._sections['Version'] = sec['Version']
finally:
del sec
@@ -490,9 +478,9 @@ class DscSrcPackage(DebPackage):
"""Check if the package is installable.."""
if not self.check_conflicts():
for pkgname in self._installed_conflicts:
- if self._cache[pkgname]._pkg.Essential:
+ if self._cache[pkgname]._pkg.essential:
raise Exception(_("An essential package would be removed"))
- self._cache[pkgname].markDelete()
+ self._cache[pkgname].mark_delete()
# FIXME: a additional run of the checkConflicts()
# after _satisfyDepends() should probably be done
return self._satisfy_depends(self.depends)
@@ -507,7 +495,7 @@ def _test():
vp = "www-browser"
#print "%s virtual: %s" % (vp, cache.isVirtualPackage(vp))
- providers = cache.getProvidingPackages(vp)
+ providers = cache.get_providing_packages(vp)
print "Providers for %s :" % vp
for pkg in providers:
print " %s" % pkg.name
@@ -525,13 +513,13 @@ def _test():
print ret
#s = DscSrcPackage(cache, "../tests/3ddesktop_0.2.9-6.dsc")
- #s.checkDep()
+ #s.check_dep()
#print "Missing deps: ",s.missingDeps
#print "Print required changes: ", s.requiredChanges
s = DscSrcPackage(cache=cache)
d = "libc6 (>= 2.3.2), libaio (>= 0.3.96) | libaio1 (>= 0.3.96)"
- print s._satisfy_depends(apt_pkg.ParseDepends(d))
+ print s._satisfy_depends(apt_pkg.parse_depends(d))
if __name__ == "__main__":
_test()
diff --git a/apt/deprecation.py b/apt/deprecation.py
new file mode 100644
index 00000000..0f39ad63
--- /dev/null
+++ b/apt/deprecation.py
@@ -0,0 +1,106 @@
+# deprecation.py - Module providing classes and functions for deprecation.
+#
+# Copyright (c) 2009 Julian Andres Klode <jak@debian.org>
+# Copyright (c) 2009 Ben Finney <ben+debian@benfinney.id.au>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+"""Classes and functions for deprecating features.
+
+This is used for internal purposes only and not part of the official API. Do
+not use it for anything outside the apt package.
+"""
+import re
+import operator
+import warnings
+
+import apt_pkg
+
+__all__ = []
+
+
+class AttributeDeprecatedBy(object):
+ """Property acting as a proxy for a new attribute.
+
+ When accessed, the property issues a DeprecationWarning and (on get) calls
+ attrgetter() for the attribute 'attribute' on the current object or (on
+ set) uses setattr to set the value of the wrapped attribute.
+ """
+
+ def __init__(self, attribute):
+ """Initialize the property."""
+ self.attribute = attribute
+ self.__doc__ = 'Deprecated, please use \'%s\' instead' % attribute
+ self.getter = operator.attrgetter(attribute)
+
+ def __get__(self, obj, type_=None):
+ """Issue a DeprecationWarning and return the requested value."""
+ if obj is None:
+ return getattr(type_, self.attribute, self)
+ warnings.warn(self.__doc__, DeprecationWarning, stacklevel=2)
+ return self.getter(obj or type_)
+
+ def __set__(self, obj, value):
+ """Issue a DeprecationWarning and set the requested value."""
+ warnings.warn(self.__doc__, DeprecationWarning, stacklevel=2)
+ setattr(obj, self.attribute, value)
+
+
+def function_deprecated_by(func, convert_names=True):
+ """Return a function that warns it is deprecated by another function.
+
+ Returns a new function that warns it is deprecated by function 'func',
+ then acts as a pass-through wrapper for 'func'.
+
+ This function also converts all keyword argument names from mixedCase to
+ lowercase_with_underscores, but only if 'convert_names' is True (default).
+ """
+ warning = 'Deprecated, please use \'%s()\' instead' % func.__name__
+
+ def deprecated_function(*args, **kwds):
+ """Wrapper around a deprecated function."""
+ warnings.warn(warning, DeprecationWarning, stacklevel=2)
+ if convert_names:
+ for key in kwds.keys():
+ kwds[re.sub('([A-Z])', '_\\1', key).lower()] = kwds.pop(key)
+ return func(*args, **kwds)
+ return deprecated_function
+
+
+def deprecated_args(func):
+ """A function with deprecated arguments.
+
+ Similar to function_deprecated_by() but warns on every deprecated argument
+ instead of function calls.
+ """
+ if not apt_pkg._COMPAT_0_7:
+ return func
+
+ def deprecated_function(*args, **kwds):
+ """Wrapper around a function with deprecated arguments."""
+ for key in kwds.keys():
+ new_key = re.sub('([A-Z])', '_\\1', key).lower()
+ if new_key != key:
+ warnings.warn("Deprecated parameter %r" % key)
+ kwds[new_key] = kwds.pop(key)
+ return func(*args, **kwds)
+
+ # Wrap the function completely (same as functools.wraps)
+ # pylint: disable-msg=W0622
+ deprecated_function.__name__ = func.__name__
+ deprecated_function.__doc__ = func.__doc__
+ deprecated_function.__module__ = func.__module__
+ deprecated_function.__dict__.update(func.__dict__)
+ return deprecated_function
diff --git a/apt/package.py b/apt/package.py
index f33871c6..7f736583 100644
--- a/apt/package.py
+++ b/apt/package.py
@@ -19,7 +19,6 @@
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
# USA
"""Functionality related to packages."""
-import gettext
import httplib
import os
import sys
@@ -28,6 +27,11 @@ import socket
import subprocess
import urllib2
import warnings
+try:
+ from collections import Mapping, Sequence
+except ImportError:
+ # (for Python < 2.6) pylint: disable-msg=C0103
+ Sequence = Mapping = object
try:
from collections import Sequence
@@ -35,19 +39,17 @@ except ImportError:
Sequence = object
import apt_pkg
-import apt.progress
+import apt.progress.text
+from apt_pkg import gettext as _
+from apt.deprecation import (function_deprecated_by, AttributeDeprecatedBy,
+ deprecated_args)
__all__ = ('BaseDependency', 'Dependency', 'Origin', 'Package', 'Record',
'Version', 'VersionList')
-def _(string):
- """Return the translation of the string."""
- return gettext.dgettext("python-apt", string)
-
-
def _file_is_same(path, size, md5):
- """Return True if the file is the same."""
+ """Return ``True`` if the file is the same."""
if (os.path.exists(path) and os.path.getsize(path) == size and
apt_pkg.md5sum(open(path)) == md5):
return True
@@ -61,23 +63,26 @@ class BaseDependency(object):
"""A single dependency.
Attributes defined here:
- name - The name of the dependency
- relation - The relation (>>,>=,==,<<,<=,)
- version - The version depended on
+ name - The name of the dependency
+ relation - The relation (>,>=,==,<,<=,)
+ version - The version depended on
rawtype - The type of the dependendy (e.g. 'Recommends')
- preDepend - Boolean value whether this is a pre-dependency.
+ pre_depend - Boolean value whether this is a pre-dependency.
"""
def __init__(self, name, rel, ver, pre, rawtype=None):
self.name = name
self.relation = rel
self.version = ver
- self.preDepend = pre
+ self.pre_depend = pre
self.rawtype = rawtype
def __repr__(self):
return ('<BaseDependency: name:%r relation:%r version:%r preDepend:%r>'
- % (self.name, self.relation, self.version, self.preDepend))
+ % (self.name, self.relation, self.version, self.pre_depend))
+
+ if apt_pkg._COMPAT_0_7:
+ preDepend = AttributeDeprecatedBy('pre_depend')
class Dependency(object):
@@ -103,15 +108,15 @@ class DeprecatedProperty(property):
def __init__(self, fget=None, fset=None, fdel=None, doc=None):
property.__init__(self, fget, fset, fdel, doc)
- self.__doc__ = ':Deprecated: ' + (doc or fget.__doc__ or '')
+ self.__doc__ = (doc or fget.__doc__ or '')
- def __get__(self, obj, type=None):
+ def __get__(self, obj, type_=None):
if obj is not None:
warnings.warn("Accessed deprecated property %s.%s, please see the "
"Version class for alternatives." %
- ((obj.__class__.__name__ or type.__name__),
- self.fget.func_name), DeprecationWarning, 2)
- return property.__get__(self, obj, type)
+ ((obj.__class__.__name__ or type_.__name__),
+ self.fget.__name__), DeprecationWarning, 2)
+ return property.__get__(self, obj, type_)
class Origin(object):
@@ -126,16 +131,16 @@ class Origin(object):
trusted - Boolean value whether this is trustworthy.
"""
- def __init__(self, pkg, VerFileIter):
- self.archive = VerFileIter.Archive
- self.component = VerFileIter.Component
- self.label = VerFileIter.Label
- self.origin = VerFileIter.Origin
- self.site = VerFileIter.Site
- self.not_automatic = VerFileIter.NotAutomatic
+ def __init__(self, pkg, packagefile):
+ self.archive = packagefile.archive
+ self.component = packagefile.component
+ self.label = packagefile.label
+ self.origin = packagefile.origin
+ self.site = packagefile.site
+ self.not_automatic = packagefile.not_automatic
# check the trust
- indexfile = pkg._pcache._list.FindIndex(VerFileIter)
- if indexfile and indexfile.IsTrusted:
+ indexfile = pkg._pcache._list.find_index(packagefile)
+ if indexfile and indexfile.is_trusted:
self.trusted = True
else:
self.trusted = False
@@ -147,7 +152,7 @@ class Origin(object):
self.site, self.trusted)
-class Record(object):
+class Record(Mapping):
"""Represent a pkgRecord.
It can be accessed like a dictionary and can also give the original package
@@ -155,7 +160,10 @@ class Record(object):
"""
def __init__(self, record_str):
- self._rec = apt_pkg.ParseSection(record_str)
+ self._rec = apt_pkg.TagSection(record_str)
+
+ def __hash__(self):
+ return hash(self._rec)
def __str__(self):
return str(self._rec)
@@ -164,7 +172,7 @@ class Record(object):
return self._rec[key]
def __contains__(self, key):
- return self._rec.has_key(key)
+ return key in self._rec
def __iter__(self):
return iter(self._rec.keys())
@@ -175,21 +183,24 @@ class Record(object):
yield key, self._rec[key]
def get(self, key, default=None):
- """Return record[key] if key in record, else `default`.
+ """Return record[key] if key in record, else *default*.
- The parameter `default` must be either a string or None.
+ The parameter *default* must be either a string or None.
"""
return self._rec.get(key, default)
def has_key(self, key):
- """deprecated form of 'key in x'."""
- return self._rec.has_key(key)
+ """deprecated form of ``key in x``."""
+ return key in self._rec
+
+ def __len__(self):
+ return len(self._rec)
class Version(object):
"""Representation of a package version.
- :since: 0.7.9
+ .. versionadded:: 0.7.9
"""
def __init__(self, package, cand):
@@ -198,9 +209,9 @@ class Version(object):
def _cmp(self, other):
try:
- return apt_pkg.VersionCompare(self._cand.VerStr, other.version)
+ return apt_pkg.version_compare(self._cand.ver_str, other.version)
except AttributeError:
- return apt_pkg.VersionCompare(self._cand.VerStr, other)
+ return apt_pkg.version_compare(self._cand.ver_str, other)
def __eq__(self, other):
try:
@@ -239,7 +250,7 @@ class Version(object):
return NotImplemented
def __hash__(self):
- return self._cand.Hash
+ return self._cand.hash
def __repr__(self):
return '<Version: package:%r version:%r>' % (self.package.name,
@@ -248,58 +259,63 @@ class Version(object):
@property
def _records(self):
"""Internal helper that moves the Records to the right position."""
- if self.package._pcache._records.Lookup(self._cand.FileList[0]):
+ if self.package._pcache._records.lookup(self._cand.file_list[0]):
return self.package._pcache._records
@property
+ def _translated_records(self):
+ """Internal helper to get the translated description."""
+ desc_iter = self._cand.translated_description
+ self.package._pcache._records.lookup(desc_iter.file_list.pop(0))
+ return self.package._pcache._records
+
+ @property
def installed_size(self):
"""Return the size of the package when installed."""
- return self._cand.InstalledSize
+ return self._cand.installed_size
@property
def homepage(self):
"""Return the homepage for the package."""
- return self._records.Homepage
+ return self._records.homepage
@property
def size(self):
"""Return the size of the package."""
- return self._cand.Size
+ return self._cand.size
@property
def architecture(self):
"""Return the architecture of the package version."""
- return self._cand.Arch
+ return self._cand.arch
@property
def downloadable(self):
"""Return whether the version of the package is downloadable."""
- return bool(self._cand.Downloadable)
+ return bool(self._cand.downloadable)
@property
def version(self):
"""Return the version as a string."""
- return self._cand.VerStr
+ return self._cand.ver_str
@property
def summary(self):
"""Return the short description (one line summary)."""
- desc_iter = self._cand.TranslatedDescription
- self.package._pcache._records.Lookup(desc_iter.FileList.pop(0))
- return self.package._pcache._records.ShortDesc
+ return self._translated_records.short_desc
@property
def raw_description(self):
"""return the long description (raw)."""
- return self._records.LongDesc
+ return self._records.long_desc
@property
def section(self):
"""Return the section of the package."""
- return self._cand.Section
+ return self._cand.section
@property
- def description(self, format=True, useDots=False):
+ def description(self):
"""Return the formatted long description.
Return the formated long description according to the Debian policy
@@ -307,10 +323,12 @@ class Version(object):
See http://www.debian.org/doc/debian-policy/ch-controlfields.html
for more information.
"""
- self.summary # This does the lookup for us.
desc = ''
+ dsc = self._translated_records.long_desc
try:
- dsc = unicode(self.package._pcache._records.LongDesc, "utf-8")
+ if not isinstance(dsc, unicode):
+ # Only convert where needed (i.e. Python 2.X)
+ dsc = unicode(dsc, "utf-8")
except UnicodeDecodeError, err:
return _("Invalid unicode in description for '%s' (%s). "
"Please report.") % (self.package.name, err)
@@ -347,32 +365,33 @@ class Version(object):
def source_name(self):
"""Return the name of the source package."""
try:
- return self._records.SourcePkg or self.package.name
+ return self._records.source_pkg or self.package.name
except IndexError:
return self.package.name
@property
def priority(self):
"""Return the priority of the package, as string."""
- return self._cand.PriorityStr
+ return self._cand.priority_str
@property
def record(self):
"""Return a Record() object for this version."""
- return Record(self._records.Record)
+ return Record(self._records.record)
def get_dependencies(self, *types):
"""Return a list of Dependency objects for the given types."""
depends_list = []
- depends = self._cand.DependsList
- for t in types:
+ depends = self._cand.depends_list
+ for type_ in types:
try:
- for depVerList in depends[t]:
+ for dep_ver_list in depends[type_]:
base_deps = []
- for depOr in depVerList:
- base_deps.append(BaseDependency(depOr.TargetPkg.Name,
- depOr.CompType, depOr.TargetVer,
- (t == "PreDepends"), rawtype=t))
+ for dep_or in dep_ver_list:
+ base_deps.append(BaseDependency(dep_or.target_pkg.name,
+ dep_or.comp_type, dep_or.target_ver,
+ (type_ == "PreDepends"),
+ rawtype=type_))
depends_list.append(Dependency(base_deps))
except KeyError:
pass
@@ -397,123 +416,144 @@ class Version(object):
def origins(self):
"""Return a list of origins for the package version."""
origins = []
- for (verFileIter, index) in self._cand.FileList:
- origins.append(Origin(self.package, verFileIter))
+ for (packagefile, _) in self._cand.file_list:
+ origins.append(Origin(self.package, packagefile))
return origins
@property
def filename(self):
- """Return the path to the file inside the archive."""
- return self._records.FileName
+ """Return the path to the file inside the archive.
+
+ .. versionadded:: 0.7.10
+ """
+ return self._records.filename
@property
def md5(self):
- """Return the md5sum of the binary."""
- return self._records.MD5Hash
+ """Return the md5sum of the binary.
+
+ .. versionadded:: 0.7.10
+ """
+ return self._records.md5_hash
@property
def sha1(self):
- """Return the sha1sum of the binary."""
- return self._records.SHA1Hash
+ """Return the sha1sum of the binary.
+
+ .. versionadded:: 0.7.10
+ """
+ return self._records.sha1_hash
@property
def sha256(self):
- """Return the sha1sum of the binary."""
- return self._records.SHA256Hash
+ """Return the sha256sum of the binary.
+
+ .. versionadded:: 0.7.10
+ """
+ return self._records.sha256_hash
def _uris(self):
- """Return an iterator over all available urls."""
- for (packagefile, index) in self._cand.FileList:
- indexfile = self.package._pcache._list.FindIndex(packagefile)
+ """Return an iterator over all available urls.
+
+ .. versionadded:: 0.7.10
+ """
+ for (packagefile, _) in self._cand.file_list:
+ indexfile = self.package._pcache._list.find_index(packagefile)
if indexfile:
- yield indexfile.ArchiveURI(self._records.FileName)
+ yield indexfile.archive_uri(self._records.filename)
@property
def uris(self):
- """Return a list of all available uris for the binary."""
+ """Return a list of all available uris for the binary.
+
+ .. versionadded:: 0.7.10
+ """
return list(self._uris())
@property
def uri(self):
- """Return a single URI for the binary."""
- return self._uris().next()
+ """Return a single URI for the binary.
+
+ .. versionadded:: 0.7.10
+ """
+ return iter(self._uris()).next()
def fetch_binary(self, destdir='', progress=None):
"""Fetch the binary version of the package.
- The parameter 'destdir' specifies the directory where the package will
+ The parameter *destdir* specifies the directory where the package will
be fetched to.
- The parameter 'progress' may refer to an apt.progress.FetchProgress()
- object. If not specified or None, apt.progress.TextFetchProgress() is
- used.
+ The parameter *progress* may refer to an apt_pkg.AcquireProgress()
+ object. If not specified or None, apt.progress.text.AcquireProgress()
+ is used.
+
+ .. versionadded:: 0.7.10
"""
- base = os.path.basename(self._records.FileName)
+ base = os.path.basename(self._records.filename)
destfile = os.path.join(destdir, base)
- if _file_is_same(destfile, self.size, self._records.MD5Hash):
+ if _file_is_same(destfile, self.size, self._records.md5_hash):
print 'Ignoring already existing file:', destfile
return
- acq = apt_pkg.GetAcquire(progress or apt.progress.TextFetchProgress())
- apt_pkg.GetPkgAcqFile(acq, self.uri, self._records.MD5Hash, self.size,
- base, destFile=destfile)
- acq.Run()
- for item in acq.Items:
- if item.Status != item.StatDone:
- raise FetchError("The item %r could not be fetched: %s" %
- (item.DestFile, item.ErrorText))
+ acq = apt_pkg.Acquire(progress or apt.progress.text.AcquireProgress())
+ acqfile = apt_pkg.AcquireFile(acq, self.uri, self._records.md5_hash,
+ self.size, base, destfile=destfile)
+ acq.run()
+
+ if acqfile.status != acqfile.STAT_DONE:
+ raise FetchError("The item %r could not be fetched: %s" %
+ (acqfile.destfile, acqfile.error_text))
+ print self._records.filename
return os.path.abspath(destfile)
def fetch_source(self, destdir="", progress=None, unpack=True):
"""Get the source code of a package.
- The parameter 'destdir' specifies the directory where the source will
+ The parameter *destdir* specifies the directory where the source will
be fetched to.
- The parameter 'progress' may refer to an apt.progress.FetchProgress()
- object. If not specified or None, apt.progress.TextFetchProgress() is
- used.
+ The parameter *progress* may refer to an apt_pkg.AcquireProgress()
+ object. If not specified or None, apt.progress.text.AcquireProgress()
+ is used.
- The parameter 'unpack' describes whether the source should be unpacked
- (True) or not (False). By default, it is unpacked.
+ The parameter *unpack* describes whether the source should be unpacked
+ (``True``) or not (``False``). By default, it is unpacked.
- If 'unpack' is True, the path to the extracted directory is returned.
- Otherwise, the path to the .dsc file is returned.
+ If *unpack* is ``True``, the path to the extracted directory is
+ returned. Otherwise, the path to the .dsc file is returned.
"""
- src = apt_pkg.GetPkgSrcRecords()
- acq = apt_pkg.GetAcquire(progress or apt.progress.TextFetchProgress())
+ src = apt_pkg.SourceRecords()
+ acq = apt_pkg.Acquire(progress or apt.progress.text.AcquireProgress())
dsc = None
record = self._records
- src.Lookup(record.SourcePkg)
+ src.lookup(record.source_pkg)
+
try:
- while record.SourceVer != src.Version:
- src.Lookup(record.SourcePkg)
+ while record.source_ver != src.version:
+ src.lookup(record.source_pkg)
except AttributeError:
raise ValueError("No source for %r" % self)
- for md5, size, path, type in src.Files:
+ files = list()
+ for md5, size, path, type_ in src.files:
base = os.path.basename(path)
destfile = os.path.join(destdir, base)
- if type == 'dsc':
+ if type_ == 'dsc':
dsc = destfile
- if os.path.exists(base) and os.path.getsize(base) == size:
- fobj = open(base)
- try:
- if apt_pkg.md5sum(fobj) == md5:
- print 'Ignoring already existing file:', destfile
- continue
- finally:
- fobj.close()
- apt_pkg.GetPkgAcqFile(acq, src.Index.ArchiveURI(path), md5, size,
- base, destFile=destfile)
- acq.Run()
-
- for item in acq.Items:
- if item.Status != item.StatDone:
+ if _file_is_same(destfile, size, md5):
+ print 'Ignoring already existing file:', destfile
+ continue
+ files.append(apt_pkg.AcquireFile(acq, src.index.archive_uri(path),
+ md5, size, base, destfile=destfile))
+ acq.run()
+
+ for item in acq.items:
+ if item.status != item.STAT_DONE:
raise FetchError("The item %r could not be fetched: %s" %
- (item.DestFile, item.ErrorText))
+ (item.destfile, item.error_text))
if unpack:
- outdir = src.Package + '-' + apt_pkg.UpstreamVersion(src.Version)
+ outdir = src.package + '-' + apt_pkg.upstream_version(src.version)
outdir = os.path.join(destdir, outdir)
subprocess.check_call(["dpkg-source", "-x", dsc, outdir])
return os.path.abspath(outdir)
@@ -541,11 +581,11 @@ class VersionList(Sequence):
max(package.versions)
"""
- def __init__(self, package, slice=None):
+ def __init__(self, package, slice_=None):
self._package = package # apt.package.Package()
- self._versions = package._pkg.VersionList # [apt_pkg.Version(), ...]
- if slice:
- self._versions = self._versions[slice]
+ self._versions = package._pkg.version_list # [apt_pkg.Version(), ...]
+ if slice_:
+ self._versions = self._versions[slice_]
def __getitem__(self, item):
if isinstance(item, slice):
@@ -556,7 +596,7 @@ class VersionList(Sequence):
except TypeError:
# Dictionary interface item is a string.
for ver in self._versions:
- if ver.VerStr == item:
+ if ver.ver_str == item:
return Version(self._package, ver)
raise KeyError("Version: %r not found." % (item))
@@ -572,7 +612,7 @@ class VersionList(Sequence):
item = item.version
# Dictionary interface.
for ver in self._versions:
- if ver.VerStr == item:
+ if ver.ver_str == item:
return True
return False
@@ -586,7 +626,7 @@ class VersionList(Sequence):
def keys(self):
"""Return a list of all versions, as strings."""
- return [ver.VerStr for ver in self._versions]
+ return [ver.ver_str for ver in self._versions]
def get(self, key, default=None):
"""Return the key or the default."""
@@ -611,7 +651,7 @@ class Package(object):
self._changelog = "" # Cached changelog
def __repr__(self):
- return '<Package: name:%r id:%r>' % (self._pkg.Name, self._pkg.ID)
+ return '<Package: name:%r id:%r>' % (self._pkg.name, self._pkg.id)
def candidate(self):
"""Return the candidate version of the package.
@@ -620,15 +660,15 @@ class Package(object):
of the package. Just assign a Version() object, and it will be set as
the candidate version.
"""
- cand = self._pcache._depcache.GetCandidateVer(self._pkg)
+ cand = self._pcache._depcache.get_candidate_ver(self._pkg)
if cand is not None:
return Version(self, cand)
def __set_candidate(self, version):
"""Set the candidate version of the package."""
- self._pcache.cachePreChange()
- self._pcache._depcache.SetCandidateVer(self._pkg, version._cand)
- self._pcache.cachePostChange()
+ self._pcache.cache_pre_change()
+ self._pcache._depcache.set_candidate_ver(self._pkg, version._cand)
+ self._pcache.cache_post_change()
candidate = property(candidate, __set_candidate)
@@ -636,104 +676,137 @@ class Package(object):
def installed(self):
"""Return the currently installed version of the package.
- :since: 0.7.9"""
- if self._pkg.CurrentVer is not None:
- return Version(self, self._pkg.CurrentVer)
+ .. versionadded:: 0.7.9
+ """
+ if self._pkg.current_ver is not None:
+ return Version(self, self._pkg.current_ver)
@property
def name(self):
"""Return the name of the package."""
- return self._pkg.Name
+ return self._pkg.name
@property
def id(self):
"""Return a uniq ID for the package.
This can be used eg. to store additional information about the pkg."""
- return self._pkg.ID
+ return self._pkg.id
def __hash__(self):
"""Return the hash of the object.
This returns the same value as ID, which is unique."""
- return self._pkg.ID
+ return self._pkg.id
@property
def essential(self):
"""Return True if the package is an essential part of the system."""
- return self._pkg.Essential
+ return self._pkg.essential
@DeprecatedProperty
- def installedVersion(self):
+ def installedVersion(self): #pylint: disable-msg=C0103
"""Return the installed version as string.
- Deprecated, please use installed.version instead."""
+ .. deprecated:: 0.7.9"""
return getattr(self.installed, 'version', None)
@DeprecatedProperty
- def candidateVersion(self):
- """Return the candidate version as string."""
+ def candidateVersion(self): #pylint: disable-msg=C0103
+ """Return the candidate version as string.
+
+ .. deprecated:: 0.7.9"""
return getattr(self.candidate, "version", None)
@DeprecatedProperty
- def candidateDependencies(self):
- """Return a list of candidate dependencies."""
+ def candidateDependencies(self): #pylint: disable-msg=C0103
+ """Return a list of candidate dependencies.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "dependencies", None)
@DeprecatedProperty
- def installedDependencies(self):
- """Return a list of installed dependencies."""
+ def installedDependencies(self): #pylint: disable-msg=C0103
+ """Return a list of installed dependencies.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.installed, 'dependencies', [])
@DeprecatedProperty
def architecture(self):
- """Return the Architecture of the package"""
+ """Return the Architecture of the package.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "architecture", None)
@DeprecatedProperty
- def candidateDownloadable(self):
- """Return True if the candidate is downloadable."""
+ def candidateDownloadable(self): #pylint: disable-msg=C0103
+ """Return ``True`` if the candidate is downloadable.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "downloadable", None)
@DeprecatedProperty
- def installedDownloadable(self):
- """Return True if the installed version is downloadable."""
+ def installedDownloadable(self): #pylint: disable-msg=C0103
+ """Return ``True`` if the installed version is downloadable.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.installed, 'downloadable', False)
@DeprecatedProperty
- def sourcePackageName(self):
- """Return the source package name as string."""
+ def sourcePackageName(self): #pylint: disable-msg=C0103
+ """Return the source package name as string.
+
+ .. deprecated:: 0.7.9
+ """
try:
- return self.candidate._records.SourcePkg or self._pkg.Name
+ return self.candidate._records.source_pkg or self._pkg.name
except AttributeError:
try:
- return self.installed._records.SourcePkg or self._pkg.Name
+ return self.installed._records.source_pkg or self._pkg.name
except AttributeError:
- return self._pkg.Name
+ return self._pkg.name
@DeprecatedProperty
def homepage(self):
- """Return the homepage field as string."""
+ """Return the homepage field as string.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "homepage", None)
@property
def section(self):
"""Return the section of the package."""
- return self._pkg.Section
+ return self._pkg.section
@DeprecatedProperty
def priority(self):
- """Return the priority (of the candidate version)."""
+ """Return the priority (of the candidate version).
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "priority", None)
@DeprecatedProperty
- def installedPriority(self):
- """Return the priority (of the installed version)."""
+ def installedPriority(self): #pylint: disable-msg=C0103
+ """Return the priority (of the installed version).
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.installed, 'priority', None)
@DeprecatedProperty
def summary(self):
- """Return the short description (one line summary)."""
+ """Return the short description (one line summary).
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "summary", None)
@DeprecatedProperty
@@ -744,101 +817,127 @@ class Package(object):
(Chapter 5.6.13).
See http://www.debian.org/doc/debian-policy/ch-controlfields.html
for more information.
+
+ .. deprecated:: 0.7.9
"""
return getattr(self.candidate, "description", None)
@DeprecatedProperty
- def rawDescription(self):
- """return the long description (raw)."""
+ def rawDescription(self): #pylint: disable-msg=C0103
+ """return the long description (raw).
+
+ .. deprecated:: 0.7.9"""
return getattr(self.candidate, "raw_description", None)
@DeprecatedProperty
- def candidateRecord(self):
- """Return the Record of the candidate version of the package."""
+ def candidateRecord(self): #pylint: disable-msg=C0103
+ """Return the Record of the candidate version of the package.
+
+ .. deprecated:: 0.7.9"""
return getattr(self.candidate, "record", None)
@DeprecatedProperty
- def installedRecord(self):
- """Return the Record of the candidate version of the package."""
+ def installedRecord(self): #pylint: disable-msg=C0103
+ """Return the Record of the candidate version of the package.
+
+ .. deprecated:: 0.7.9"""
return getattr(self.installed, 'record', '')
# depcache states
@property
- def markedInstall(self):
- """Return True if the package is marked for install."""
- return self._pcache._depcache.MarkedInstall(self._pkg)
+ def marked_install(self):
+ """Return ``True`` if the package is marked for install."""
+ return self._pcache._depcache.marked_install(self._pkg)
@property
- def markedUpgrade(self):
- """Return True if the package is marked for upgrade."""
- return self._pcache._depcache.MarkedUpgrade(self._pkg)
+ def marked_upgrade(self):
+ """Return ``True`` if the package is marked for upgrade."""
+ return self._pcache._depcache.marked_upgrade(self._pkg)
@property
- def markedDelete(self):
- """Return True if the package is marked for delete."""
- return self._pcache._depcache.MarkedDelete(self._pkg)
+ def marked_delete(self):
+ """Return ``True`` if the package is marked for delete."""
+ return self._pcache._depcache.marked_delete(self._pkg)
@property
- def markedKeep(self):
- """Return True if the package is marked for keep."""
- return self._pcache._depcache.MarkedKeep(self._pkg)
+ def marked_keep(self):
+ """Return ``True`` if the package is marked for keep."""
+ return self._pcache._depcache.marked_keep(self._pkg)
@property
- def markedDowngrade(self):
+ def marked_downgrade(self):
""" Package is marked for downgrade """
- return self._pcache._depcache.MarkedDowngrade(self._pkg)
+ return self._pcache._depcache.marked_downgrade(self._pkg)
@property
- def markedReinstall(self):
- """Return True if the package is marked for reinstall."""
- return self._pcache._depcache.MarkedReinstall(self._pkg)
+ def marked_reinstall(self):
+ """Return ``True`` if the package is marked for reinstall."""
+ return self._pcache._depcache.marked_reinstall(self._pkg)
@property
- def isInstalled(self):
- """Return True if the package is installed."""
- return (self._pkg.CurrentVer is not None)
+ def is_installed(self):
+ """Return ``True`` if the package is installed."""
+ return (self._pkg.current_ver is not None)
@property
- def isUpgradable(self):
- """Return True if the package is upgradable."""
- return (self.isInstalled and
- self._pcache._depcache.IsUpgradable(self._pkg))
+ def is_upgradable(self):
+ """Return ``True`` if the package is upgradable."""
+ return (self.is_installed and
+ self._pcache._depcache.is_upgradable(self._pkg))
@property
- def isAutoRemovable(self):
- """Return True if the package is no longer required.
+ def is_auto_removable(self):
+ """Return ``True`` if the package is no longer required.
If the package has been installed automatically as a dependency of
another package, and if no packages depend on it anymore, the package
is no longer required.
"""
- return self.isInstalled and self._pcache._depcache.IsGarbage(self._pkg)
+ return self.is_installed and \
+ self._pcache._depcache.is_garbage(self._pkg)
+ @property
+ def is_auto_installed(self):
+ """Return whether the package is marked as automatically installed."""
+ return self._pcache._depcache.is_auto_installed(self._pkg)
# sizes
@DeprecatedProperty
- def packageSize(self):
- """Return the size of the candidate deb package."""
+ def packageSize(self): #pylint: disable-msg=C0103
+ """Return the size of the candidate deb package.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "size", None)
@DeprecatedProperty
- def installedPackageSize(self):
- """Return the size of the installed deb package."""
+ def installedPackageSize(self): #pylint: disable-msg=C0103
+ """Return the size of the installed deb package.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.installed, 'size', 0)
@DeprecatedProperty
- def candidateInstalledSize(self):
- """Return the size of the candidate installed package."""
+ def candidateInstalledSize(self): #pylint: disable-msg=C0103
+ """Return the size of the candidate installed package.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "installed_size", None)
@DeprecatedProperty
- def installedSize(self):
- """Return the size of the currently installed package."""
+ def installedSize(self): #pylint: disable-msg=C0103
+ """Return the size of the currently installed package.
+
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.installed, 'installed_size', 0)
@property
- def installedFiles(self):
+ def installed_files(self):
"""Return a list of files installed by the package.
Return a list of unicode names of the files which have
@@ -854,20 +953,21 @@ class Package(object):
except EnvironmentError:
return []
- def getChangelog(self, uri=None, cancel_lock=None):
+ def get_changelog(self, uri=None, cancel_lock=None):
"""
Download the changelog of the package and return it as unicode
string.
- The parameter `uri` refers to the uri of the changelog file. It may
+ The parameter *uri* refers to the uri of the changelog file. It may
contain multiple named variables which will be substitued. These
variables are (src_section, prefix, src_pkg, src_ver). An example is
- the Ubuntu changelog:
+ the Ubuntu changelog::
+
"http://changelogs.ubuntu.com/changelogs/pool" \\
"/%(src_section)s/%(prefix)s/%(src_pkg)s" \\
"/%(src_pkg)s_%(src_ver)s/changelog"
- The parameter `cancel_lock` refers to an instance of threading.Lock,
+ The parameter *cancel_lock* refers to an instance of threading.Lock,
which if set, prevents the download.
"""
# Return a cached changelog if available
@@ -907,16 +1007,16 @@ class Package(object):
# this feature only works if the correct deb-src are in the
# sources.list
# otherwise we fall back to the binary version number
- src_records = apt_pkg.GetPkgSrcRecords()
- src_rec = src_records.Lookup(src_pkg)
+ src_records = apt_pkg.SourceRecords()
+ src_rec = src_records.lookup(src_pkg)
if src_rec:
- src_ver = src_records.Version
+ src_ver = src_records.version
#if apt_pkg.VersionCompare(binver, srcver) > 0:
# srcver = binver
if not src_ver:
src_ver = bin_ver
#print "srcver: %s" % src_ver
- section = src_records.Section
+ section = src_records.section
#print "srcsect: %s" % section
else:
# fail into the error handler
@@ -924,9 +1024,10 @@ class Package(object):
except SystemError:
src_ver = bin_ver
- l = section.split("/")
- if len(l) > 1:
- src_section = l[0]
+ section_split = section.split("/", 1)
+ if len(section_split) > 1:
+ src_section = section_split[0]
+ del section_split
# lib is handled special
prefix = src_pkg[0]
@@ -934,9 +1035,10 @@ class Package(object):
prefix = "lib" + src_pkg[3]
# stip epoch
- l = src_ver.split(":")
- if len(l) > 1:
- src_ver = "".join(l[1:])
+ src_ver_split = src_ver.split(":", 1)
+ if len(src_ver_split) > 1:
+ src_ver = "".join(src_ver_split[1:])
+ del src_ver_split
uri = uri % {"src_section": src_section,
"prefix": prefix,
@@ -976,14 +1078,14 @@ class Package(object):
if match:
# strip epoch from installed version
# and from changelog too
- installed = self.installedVersion
+ installed = getattr(self.installed, 'version', None)
if installed and ":" in installed:
installed = installed.split(":", 1)[1]
changelog_ver = match.group(1)
if changelog_ver and ":" in changelog_ver:
changelog_ver = changelog_ver.split(":", 1)[1]
- if (installed and apt_pkg.VersionCompare(changelog_ver,
- installed) <= 0):
+ if (installed and apt_pkg.version_compare(
+ changelog_ver, installed) <= 0):
break
# EOF (shouldn't really happen)
changelog += line
@@ -1007,110 +1109,164 @@ class Package(object):
return self._changelog
@DeprecatedProperty
- def candidateOrigin(self):
- """Return a list of Origin() objects for the candidate version."""
+ def candidateOrigin(self): #pylint: disable-msg=C0103
+ """Return a list of `Origin()` objects for the candidate version.
+
+ .. deprecated:: 0.7.9
+ """
return getattr(self.candidate, "origins", None)
@property
def versions(self):
"""Return a VersionList() object for all available versions.
- :since: 0.7.9
+ .. versionadded:: 0.7.9
"""
return VersionList(self)
@property
def is_inst_broken(self):
"""Return True if the to-be-installed package is broken."""
- return self._pcache._depcache.IsInstBroken(self._pkg)
+ return self._pcache._depcache.is_inst_broken(self._pkg)
@property
def is_now_broken(self):
"""Return True if the installed package is broken."""
- return self._pcache._depcache.IsNowBroken(self._pkg)
+ return self._pcache._depcache.is_now_broken(self._pkg)
# depcache actions
- def markKeep(self):
+ def mark_keep(self):
"""Mark a package for keep."""
- self._pcache.cachePreChange()
- self._pcache._depcache.MarkKeep(self._pkg)
- self._pcache.cachePostChange()
+ self._pcache.cache_pre_change()
+ self._pcache._depcache.mark_keep(self._pkg)
+ self._pcache.cache_post_change()
- def markDelete(self, autoFix=True, purge=False):
+ @deprecated_args
+ def mark_delete(self, auto_fix=True, purge=False):
"""Mark a package for install.
- If autoFix is True, the resolver will be run, trying to fix broken
- packages. This is the default.
+ If *auto_fix* is ``True``, the resolver will be run, trying to fix
+ broken packages. This is the default.
- If purge is True, remove the configuration files of the package as
- well. The default is to keep the configuration.
+ If *purge* is ``True``, remove the configuration files of the package
+ as well. The default is to keep the configuration.
"""
- self._pcache.cachePreChange()
- self._pcache._depcache.MarkDelete(self._pkg, purge)
+ self._pcache.cache_pre_change()
+ self._pcache._depcache.mark_delete(self._pkg, purge)
# try to fix broken stuffsta
- if autoFix and self._pcache._depcache.BrokenCount > 0:
- Fix = apt_pkg.GetPkgProblemResolver(self._pcache._depcache)
- Fix.Clear(self._pkg)
- Fix.Protect(self._pkg)
- Fix.Remove(self._pkg)
- Fix.InstallProtect()
- Fix.Resolve()
- self._pcache.cachePostChange()
-
- def markInstall(self, autoFix=True, autoInst=True, fromUser=True):
+ if auto_fix and self._pcache._depcache.broken_count > 0:
+ fix = apt_pkg.ProblemResolver(self._pcache._depcache)
+ fix.clear(self._pkg)
+ fix.protect(self._pkg)
+ fix.remove(self._pkg)
+ fix.install_protect()
+ fix.resolve()
+ self._pcache.cache_post_change()
+
+ @deprecated_args
+ def mark_install(self, auto_fix=True, auto_inst=True, from_user=True):
"""Mark a package for install.
- If autoFix is True, the resolver will be run, trying to fix broken
- packages. This is the default.
+ If *autoFix* is ``True``, the resolver will be run, trying to fix
+ broken packages. This is the default.
- If autoInst is True, the dependencies of the packages will be installed
- automatically. This is the default.
+ If *autoInst* is ``True``, the dependencies of the packages will be
+ installed automatically. This is the default.
- If fromUser is True, this package will not be marked as automatically
- installed. This is the default. Set it to False if you want to be able
- to remove the package at a later stage if no other package depends on
- it.
+ If *fromUser* is ``True``, this package will not be marked as
+ automatically installed. This is the default. Set it to False if you
+ want to be able to automatically remove the package at a later stage
+ when no other package depends on it.
"""
- self._pcache.cachePreChange()
- self._pcache._depcache.MarkInstall(self._pkg, autoInst, fromUser)
+ self._pcache.cache_pre_change()
+ self._pcache._depcache.mark_install(self._pkg, auto_inst, from_user)
# try to fix broken stuff
- if autoFix and self._pcache._depcache.BrokenCount > 0:
- fixer = apt_pkg.GetPkgProblemResolver(self._pcache._depcache)
- fixer.Clear(self._pkg)
- fixer.Protect(self._pkg)
- fixer.Resolve(True)
- self._pcache.cachePostChange()
-
- def markUpgrade(self):
+ if auto_fix and self._pcache._depcache.broken_count > 0:
+ fixer = apt_pkg.ProblemResolver(self._pcache._depcache)
+ fixer.clear(self._pkg)
+ fixer.protect(self._pkg)
+ fixer.resolve(True)
+ self._pcache.cache_post_change()
+
+ def mark_upgrade(self):
"""Mark a package for upgrade."""
- if self.isUpgradable:
- fromUser = not self._pcache._depcache.IsAutoInstalled(self._pkg)
- self.markInstall(fromUser=fromUser)
+ if self.is_upgradable:
+ from_user = not self._pcache._depcache.is_auto_installed(self._pkg)
+ self.mark_install(from_user=from_user)
else:
# FIXME: we may want to throw a exception here
sys.stderr.write(("MarkUpgrade() called on a non-upgrable pkg: "
- "'%s'\n") % self._pkg.Name)
+ "'%s'\n") % self._pkg.name)
+
+ def mark_auto(self, auto=True):
+ """Mark a package as automatically installed.
+
+ Call this function to mark a package as automatically installed. If the
+ optional parameter *auto* is set to ``False``, the package will not be
+ marked as automatically installed anymore. The default is ``True``.
+ """
+ self._pcache._depcache.mark_auto(self._pkg, auto)
def commit(self, fprogress, iprogress):
"""Commit the changes.
- The parameter `fprogress` refers to a FetchProgress() object, as
- found in apt.progress.
+ The parameter *fprogress* refers to a apt_pkg.AcquireProgress() object,
+ like apt.progress.text.AcquireProgress().
- The parameter `iprogress` refers to an InstallProgress() object, as
- found in apt.progress.
+ The parameter *iprogress* refers to an InstallProgress() object, as
+ found in apt.progress.base.
"""
- self._pcache._depcache.Commit(fprogress, iprogress)
+ self._pcache._depcache.commit(fprogress, iprogress)
+
+
+ if not apt_pkg._COMPAT_0_7:
+ del installedVersion
+ del candidateVersion
+ del candidateDependencies
+ del installedDependencies
+ del architecture
+ del candidateDownloadable
+ del installedDownloadable
+ del sourcePackageName
+ del homepage
+ del priority
+ del installedPriority
+ del summary
+ del description
+ del rawDescription
+ del candidateRecord
+ del installedRecord
+ del packageSize
+ del installedPackageSize
+ del candidateInstalledSize
+ del installedSize
+ del candidateOrigin
+ else:
+ markedInstalled = AttributeDeprecatedBy('marked_installed')
+ markedInstall = AttributeDeprecatedBy('marked_install')
+ markedUpgrade = AttributeDeprecatedBy('marked_upgrade')
+ markedDelete = AttributeDeprecatedBy('marked_delete')
+ markedKeep = AttributeDeprecatedBy('marked_keep')
+ markedDowngrade = AttributeDeprecatedBy('marked_downgrade')
+ markedReinstall = AttributeDeprecatedBy('marked_reinstall')
+ isInstalled = AttributeDeprecatedBy('is_installed')
+ isUpgradable = AttributeDeprecatedBy('is_upgradable')
+ isAutoRemovable = AttributeDeprecatedBy('is_auto_removable')
+ installedFiles = AttributeDeprecatedBy('installed_files')
+ getChangelog = function_deprecated_by(get_changelog)
+ markDelete = function_deprecated_by(mark_delete)
+ markInstall = function_deprecated_by(mark_install)
+ markKeep = function_deprecated_by(mark_keep)
+ markUpgrade = function_deprecated_by(mark_upgrade)
def _test():
"""Self-test."""
print "Self-test for the Package modul"
import random
- import apt
apt_pkg.init()
- progress = apt.progress.OpTextProgress()
+ progress = apt.progress.text.OpProgress()
cache = apt.Cache(progress)
pkg = cache["apt-utils"]
print "Name: %s " % pkg.name
@@ -1132,21 +1288,21 @@ def _test():
print "Recommends: %s" % pkg.installed.recommends
for dep in pkg.candidate.dependencies:
print ",".join("%s (%s) (%s) (%s)" % (o.name, o.version, o.relation,
- o.preDepend) for o in dep.or_dependencies)
+ o.pre_depend) for o in dep.or_dependencies)
print "arch: %s" % pkg.candidate.architecture
print "homepage: %s" % pkg.candidate.homepage
print "rec: ", pkg.candidate.record
- print cache["2vcard"].getChangelog()
+ print cache["2vcard"].get_changelog()
for i in True, False:
print "Running install on random upgradable pkgs with AutoFix: %s " % i
for pkg in cache:
- if pkg.isUpgradable:
+ if pkg.is_upgradable:
if random.randint(0, 1) == 1:
- pkg.markInstall(i)
- print "Broken: %s " % cache._depcache.BrokenCount
- print "InstCount: %s " % cache._depcache.InstCount
+ pkg.mark_install(i)
+ print "Broken: %s " % cache._depcache.broken_count
+ print "InstCount: %s " % cache._depcache.inst_count
print
# get a new cache
@@ -1156,11 +1312,11 @@ def _test():
for name in cache.keys():
if random.randint(0, 1) == 1:
try:
- cache[name].markDelete(i)
+ cache[name].mark_delete(i)
except SystemError:
print "Error trying to remove: %s " % name
- print "Broken: %s " % cache._depcache.BrokenCount
- print "DelCount: %s " % cache._depcache.DelCount
+ print "Broken: %s " % cache._depcache.broken_count
+ print "DelCount: %s " % cache._depcache.del_count
# self-test
if __name__ == "__main__":
diff --git a/apt/progress/__init__.py b/apt/progress/__init__.py
index 8694de77..10c11021 100644
--- a/apt/progress/__init__.py
+++ b/apt/progress/__init__.py
@@ -1,405 +1,35 @@
-# progress.py - progress reporting classes
+# apt/progress/__init__.py - Initialization file for apt.progress.
#
-# Copyright (c) 2005-2009 Canonical
+# Copyright (c) 2009 Julian Andres Klode <jak@debian.org>
#
-# Author: Michael Vogt <michael.vogt@ubuntu.com>
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
#
-# This program is free software; you can redistribute it and/or
-# modify it under the terms of the GNU General Public License as
-# published by the Free Software Foundation; either version 2 of the
-# License, or (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU General Public License for more details.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
-# USA
-"""progress reporting classes.
-
-This module provides classes for progress reporting. They can be used with
-e.g., for reporting progress on the cache opening process, the cache update
-progress, or the package install progress.
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+"""Progress reporting.
+
+This package provides progress reporting for the python-apt package. The module
+'base' provides classes with no output, the module 'gtk2' provides classes for
+GTK+ applications, and the module 'text' provides classes for terminals, etc.
"""
-
-import errno
-import fcntl
-import os
-import re
-import select
-import sys
-
import apt_pkg
-__all__ = ('CdromProgress', 'DpkgInstallProgress', 'DumbInstallProgress',
- 'FetchProgress', 'InstallProgress', 'OpProgress', 'OpTextProgress',
- 'TextFetchProgress')
-
-
-class OpProgress(object):
- """Abstract class to implement reporting on cache opening.
-
- Subclass this class to implement simple Operation progress reporting.
- """
-
- def __init__(self):
- self.op = None
- self.subOp = None
-
- def update(self, percent):
- """Called periodically to update the user interface."""
-
- def done(self):
- """Called once an operation has been completed."""
-
-
-class OpTextProgress(OpProgress):
- """A simple text based cache open reporting class."""
-
- def __init__(self):
- OpProgress.__init__(self)
-
- def update(self, percent):
- """Called periodically to update the user interface."""
- sys.stdout.write("\r%s: %.2i " % (self.subOp, percent))
- sys.stdout.flush()
-
- def done(self):
- """Called once an operation has been completed."""
- sys.stdout.write("\r%s: Done\n" % self.op)
-
-
-class FetchProgress(object):
- """Report the download/fetching progress.
-
- Subclass this class to implement fetch progress reporting
- """
-
- # download status constants
- dlDone = 0
- dlQueued = 1
- dlFailed = 2
- dlHit = 3
- dlIgnored = 4
- dlStatusStr = {dlDone: "Done",
- dlQueued: "Queued",
- dlFailed: "Failed",
- dlHit: "Hit",
- dlIgnored: "Ignored"}
-
- def __init__(self):
- self.eta = 0.0
- self.percent = 0.0
- # Make checking easier
- self.currentBytes = 0
- self.currentItems = 0
- self.totalBytes = 0
- self.totalItems = 0
- self.currentCPS = 0
-
- def start(self):
- """Called when the fetching starts."""
-
- def stop(self):
- """Called when all files have been fetched."""
-
- def updateStatus(self, uri, descr, shortDescr, status):
- """Called when the status of an item changes.
-
- This happens eg. when the downloads fails or is completed.
- """
- def update_status_full(self, uri, descr, short_descr, status, file_size,
- partial_size):
- """Called when the status of an item changes.
-
- This happens eg. when the downloads fails or is completed. This
- version include information on current filesize and partial size
- """
-
- def pulse(self):
- """Called periodically to update the user interface.
-
- Return True to continue or False to cancel.
- """
- self.percent = (((self.currentBytes + self.currentItems) * 100.0) /
- float(self.totalBytes + self.totalItems))
- if self.currentCPS > 0:
- self.eta = ((self.totalBytes - self.currentBytes) /
- float(self.currentCPS))
- return True
-
- def pulse_items(self, items):
- """Called periodically to update the user interface.
- This function includes details about the items being fetched
- Return True to continue or False to cancel.
-
- """
- self.percent = (((self.currentBytes + self.currentItems) * 100.0) /
- float(self.totalBytes + self.totalItems))
- if self.currentCPS > 0:
- self.eta = ((self.totalBytes - self.currentBytes) /
- float(self.currentCPS))
- return True
-
- def mediaChange(self, medium, drive):
- """react to media change events."""
-
-
-class TextFetchProgress(FetchProgress):
- """ Ready to use progress object for terminal windows """
-
- def __init__(self):
- FetchProgress.__init__(self)
- self.items = {}
-
- def updateStatus(self, uri, descr, shortDescr, status):
- """Called when the status of an item changes.
-
- This happens eg. when the downloads fails or is completed.
- """
- if status != self.dlQueued:
- print "\r%s %s" % (self.dlStatusStr[status], descr)
- self.items[uri] = status
-
- def pulse(self):
- """Called periodically to update the user interface.
-
- Return True to continue or False to cancel.
- """
- FetchProgress.pulse(self)
- if self.currentCPS > 0:
- s = "[%2.f%%] %sB/s %s" % (self.percent,
- apt_pkg.SizeToStr(int(self.currentCPS)),
- apt_pkg.TimeToStr(int(self.eta)))
- else:
- s = "%2.f%% [Working]" % (self.percent)
- print "\r%s" % (s),
- sys.stdout.flush()
- return True
-
- def stop(self):
- """Called when all files have been fetched."""
- print "\rDone downloading "
-
- def mediaChange(self, medium, drive):
- """react to media change events."""
- print ("Media change: please insert the disc labeled "
- "'%s' in the drive '%s' and press enter") % (medium, drive)
-
- return raw_input() not in ('c', 'C')
-
-
-class DumbInstallProgress(object):
- """Report the install progress.
-
- Subclass this class to implement install progress reporting.
- """
-
- def startUpdate(self):
- """Start update."""
-
- def run(self, pm):
- """Start installation."""
- return pm.DoInstall()
-
- def finishUpdate(self):
- """Called when update has finished."""
-
- def updateInterface(self):
- """Called periodically to update the user interface"""
-
-
-class InstallProgress(DumbInstallProgress):
- """An InstallProgress that is pretty useful.
-
- It supports the attributes 'percent' 'status' and callbacks for the dpkg
- errors and conffiles and status changes.
- """
-
- def __init__(self):
- DumbInstallProgress.__init__(self)
- self.selectTimeout = 0.1
- (read, write) = os.pipe()
- self.writefd = write
- self.statusfd = os.fdopen(read, "r")
- fcntl.fcntl(self.statusfd.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
- self.read = ""
- self.percent = 0.0
- self.status = ""
-
- def error(self, pkg, errormsg):
- """Called when a error is detected during the install."""
-
- def conffile(self, current, new):
- """Called when a conffile question from dpkg is detected."""
-
- def statusChange(self, pkg, percent, status):
- """Called when the status changed."""
-
- def updateInterface(self):
- """Called periodically to update the interface."""
- if self.statusfd is None:
- return
- try:
- while not self.read.endswith("\n"):
- r = os.read(self.statusfd.fileno(), 1)
- if not r:
- return
- self.read += r
- except OSError, (errno_, errstr):
- # resource temporarly unavailable is ignored
- if errno_ != errno.EAGAIN and errno_ != errno.EWOULDBLOCK:
- print errstr
- if not self.read.endswith("\n"):
- return
-
- s = self.read
- #print s
- try:
- (status, pkg, percent, status_str) = s.split(":", 3)
- except ValueError:
- # silently ignore lines that can't be parsed
- self.read = ""
- return
- #print "percent: %s %s" % (pkg, float(percent)/100.0)
- if status == "pmerror":
- self.error(pkg, status_str)
- elif status == "pmconffile":
- # we get a string like this:
- # 'current-conffile' 'new-conffile' useredited distedited
- match = re.match("\s*\'(.*)\'\s*\'(.*)\'.*", status_str)
- if match:
- self.conffile(match.group(1), match.group(2))
- elif status == "pmstatus":
- if float(percent) != self.percent or status_str != self.status:
- self.statusChange(pkg, float(percent),
- status_str.strip())
- self.percent = float(percent)
- self.status = status_str.strip()
- self.read = ""
-
- def fork(self):
- """Fork."""
- return os.fork()
-
- def waitChild(self):
- """Wait for child progress to exit.
-
- The return values is the full status returned from os.waitpid()
- (not only the return code).
- """
- while True:
- try:
- select.select([self.statusfd], [], [], self.selectTimeout)
- except select.error, (errno_, errstr):
- if errno_ != errno.EINTR:
- raise
- self.updateInterface()
- try:
- (pid, res) = os.waitpid(self.child_pid, os.WNOHANG)
- if pid == self.child_pid:
- break
- except OSError, (errno_, errstr):
- if errno_ == errno.ECHILD:
- break
- if errno_ != errno.EINTR:
- raise
- return res
-
- def run(self, pm):
- """Start installing.
-
- Returns the PackageManager status:
- (pm.ResultCompleted, pm.ResultFailed, pm.ResultIncomplete)
- """
- pid = self.fork()
- if pid == 0:
- # pm.DoInstall might raise a exception,
- # when this happens, we need to catch
- # it, otherwise os._exit() is not run
- # and the execution continues in the
- # parent code leading to very confusing bugs
- try:
- res = pm.DoInstall(self.writefd)
- except Exception, e:
- os._exit(pm.ResultFailed)
- os._exit(res)
- self.child_pid = pid
- res = self.waitChild()
- return os.WEXITSTATUS(res)
-
-
-class CdromProgress(object):
- """Report the cdrom add progress.
-
- Subclass this class to implement cdrom add progress reporting.
- """
-
- def __init__(self):
- pass
-
- def update(self, text, step):
- """Called periodically to update the user interface."""
-
- def askCdromName(self):
- """Called to ask for the name of the cdrom."""
-
- def changeCdrom(self):
- """Called to ask for the cdrom to be changed."""
-
-
-class DpkgInstallProgress(InstallProgress):
- """Progress handler for a local Debian package installation."""
-
- def run(self, debfile):
- """Start installing the given Debian package."""
- self.debfile = debfile
- self.debname = os.path.basename(debfile).split("_")[0]
- pid = self.fork()
- if pid == 0:
- # child
- res = os.system("/usr/bin/dpkg --status-fd %s -i %s" % \
- (self.writefd, debfile))
- os._exit(os.WEXITSTATUS(res))
- self.child_pid = pid
- res = self.waitChild()
- return res
+__all__ = []
- def updateInterface(self):
- """Process status messages from dpkg."""
- if self.statusfd is None:
- return
- while True:
- try:
- self.read += os.read(self.statusfd.fileno(), 1)
- except OSError, (errno_, errstr):
- # resource temporarly unavailable is ignored
- if errno_ != 11:
- print errstr
- break
- if not self.read.endswith("\n"):
- continue
- statusl = self.read.split(":")
- if len(statusl) < 3:
- print "got garbage from dpkg: '%s'" % self.read
- self.read = ""
- break
- pkg_name = statusl[1].strip()
- status = statusl[2].strip()
- #print status
- if status == "error":
- self.error(pkg_name, status)
- elif status == "conffile-prompt":
- # we get a string like this:
- # 'current-conffile' 'new-conffile' useredited distedited
- match = re.match("\s*\'(.*)\'\s*\'(.*)\'.*", statusl[3])
- if match:
- self.conffile(match.group(1), match.group(2))
- else:
- self.status = status
- self.read = ""
+if apt_pkg._COMPAT_0_7:
+ from apt.progress.old import (CdromProgress, DpkgInstallProgress,
+ DumbInstallProgress, FetchProgress,
+ InstallProgress, OpProgress,
+ OpTextProgress, TextFetchProgress)
diff --git a/apt/progress/base.py b/apt/progress/base.py
new file mode 100644
index 00000000..6636cccc
--- /dev/null
+++ b/apt/progress/base.py
@@ -0,0 +1,302 @@
+# apt/progress/base.py - Base classes for progress reporting.
+#
+# Copyright (C) 2009 Julian Andres Klode <jak@debian.org>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+"""Base classes for progress reporting.
+
+Custom progress classes should inherit from these classes. They can also be
+used as dummy progress classes which simply do nothing.
+"""
+import errno
+import fcntl
+import os
+import re
+import select
+
+import apt_pkg
+
+__all__ = ['AcquireProgress', 'CdromProgress', 'InstallProgress', 'OpProgress']
+
+
+class AcquireProgress(object):
+ """Monitor object for downloads controlled by the Acquire class.
+
+ This is an mostly abstract class. You should subclass it and implement the
+ methods to get something useful.
+ """
+
+ current_bytes = current_cps = fetched_bytes = last_bytes = total_bytes \
+ = 0.0
+ current_items = elapsed_time = total_items = 0
+
+ def done(self, item):
+ """Invoked when an item is successfully and completely fetched."""
+
+ def fail(self, item):
+ """Invoked when an item could not be fetched."""
+
+ def fetch(self, item):
+ """Invoked when some of the item's data is fetched."""
+
+ def ims_hit(self, item):
+ """Invoked when an item is confirmed to be up-to-date.
+
+ Invoked when an item is confirmed to be up-to-date. For instance,
+ when an HTTP download is informed that the file on the server was
+ not modified.
+ """
+
+ def media_change(self, media, drive):
+ """Prompt the user to change the inserted removable media.
+
+ The parameter 'media' decribes the name of the the media type that
+ should be changed, whereas the parameter 'drive' should be the
+ identifying name of the drive whose media should be changed.
+
+ This method should not return until the user has confirmed to the user
+ interface that the media change is complete. It must return True if
+ the user confirms the media change, or False to cancel it.
+ """
+ return False
+
+ def pulse(self, owner):
+ """Periodically invoked while the Acquire process is underway.
+
+ This method gets invoked while the Acquire progress given by the
+ parameter 'owner' is underway. It should display information about
+ the current state.
+
+ This function returns a boolean value indicating whether the
+ acquisition should be continued (True) or cancelled (False).
+ """
+ return True
+
+ def start(self):
+ """Invoked when the Acquire process starts running."""
+ # Reset all our values.
+ self.current_bytes = 0.0
+ self.current_cps = 0.0
+ self.current_items = 0
+ self.elapsed_time = 0
+ self.fetched_bytes = 0.0
+ self.last_bytes = 0.0
+ self.total_bytes = 0.0
+ self.total_items = 0
+
+ def stop(self):
+ """Invoked when the Acquire process stops running."""
+
+
+class CdromProgress(object):
+ """Base class for reporting the progress of adding a cdrom.
+
+ Can be used with apt_pkg.Cdrom to produce an utility like apt-cdrom. The
+ attribute 'total_steps' defines the total number of steps and can be used
+ in update() to display the current progress.
+ """
+
+ total_steps = 0
+
+ def ask_cdrom_name(self):
+ """Ask for the name of the cdrom.
+
+ If a name has been provided, return it. Otherwise, return None to
+ cancel the operation.
+ """
+
+ def change_cdrom(self):
+ """Ask for the CD-ROM to be changed.
+
+ Return True once the cdrom has been changed or False to cancel the
+ operation.
+ """
+
+ def update(self, text, current):
+ """Periodically invoked to update the interface.
+
+ The string 'text' defines the text which should be displayed. The
+ integer 'current' defines the number of completed steps.
+ """
+
+
+class InstallProgress(object):
+ """Class to report the progress of installing packages."""
+
+ percent, select_timeout, status = 0.0, 0.1, ""
+
+ def __init__(self):
+ (read, write) = os.pipe()
+ self.writefd = os.fdopen(write, "w")
+ self.statusfd = os.fdopen(read, "r")
+ fcntl.fcntl(self.statusfd, fcntl.F_SETFL, os.O_NONBLOCK)
+
+ def start_update(self):
+ """(Abstract) Start update."""
+
+ def finish_update(self):
+ """(Abstract) Called when update has finished."""
+
+ def error(self, pkg, errormsg):
+ """(Abstract) Called when a error is detected during the install."""
+
+ def conffile(self, current, new):
+ """(Abstract) Called when a conffile question from dpkg is detected."""
+
+ def status_change(self, pkg, percent, status):
+ """(Abstract) Called when the APT status changed."""
+
+ def dpkg_status_change(self, pkg, status):
+ """(Abstract) Called when the dpkg status changed."""
+
+ def processing(self, pkg, stage):
+ """(Abstract) Sent just before a processing stage starts.
+
+ The parameter 'stage' is one of "upgrade", "install"
+ (both sent before unpacking), "configure", "trigproc", "remove",
+ "purge". This method is used for dpkg only.
+ """
+
+ def run(self, obj):
+ """Install using the object 'obj'.
+
+ This functions runs install actions. The parameter 'obj' may either
+ be a PackageManager object in which case its do_install() method is
+ called or the path to a deb file.
+
+ If the object is a PackageManager, the functions returns the result
+ of calling its do_install() method. Otherwise, the function returns
+ the exit status of dpkg. In both cases, 0 means that there were no
+ problems.
+ """
+ pid = self.fork()
+ if pid == 0:
+ # pm.do_install might raise a exception,
+ # when this happens, we need to catch
+ # it, otherwise os._exit() is not run
+ # and the execution continues in the
+ # parent code leading to very confusing bugs
+ try:
+ os._exit(obj.do_install(self.writefd.fileno()))
+ except AttributeError:
+ os._exit(os.spawnlp(os.P_WAIT, "dpkg", "dpkg", "--status-fd",
+ str(self.writefd.fileno()), "-i", obj))
+ except Exception:
+ os._exit(apt_pkg.PackageManager.RESULT_FAILED)
+
+ self.child_pid = pid
+ res = self.wait_child()
+ return os.WEXITSTATUS(res)
+
+ def fork(self):
+ """Fork."""
+ return os.fork()
+
+ def update_interface(self):
+ """Update the interface."""
+ try:
+ line = self.statusfd.readline()
+ except IOError, err:
+ # resource temporarly unavailable is ignored
+ if err.errno != errno.EAGAIN and err.errno != errno.EWOULDBLOCK:
+ print err.strerror
+ return
+
+ pkgname = status = status_str = percent = base = ""
+
+ if line.startswith('pm'):
+ try:
+ (status, pkgname, percent, status_str) = line.split(":", 3)
+ except ValueError:
+ # silently ignore lines that can't be parsed
+ self.read = ""
+ return
+ elif line.startswith('status'):
+ try:
+ (base, pkgname, status, status_str) = line.split(":", 3)
+ except ValueError:
+ (base, pkgname, status) = line.split(":", 2)
+ elif line.startswith('processing'):
+ (status, status_str, pkgname) = line.split(":", 2)
+ self.processing(pkgname.strip(), status_str.strip())
+
+ # Always strip the status message
+ pkgname = pkgname.strip()
+ status_str = status_str.strip()
+ status = status.strip()
+
+ if status == 'pmerror' or status == 'error':
+ self.error(pkgname, status_str)
+ elif status == 'conffile-prompt' or status == 'pmconffile':
+ match = re.match("\s*\'(.*)\'\s*\'(.*)\'.*", status_str)
+ if match:
+ self.conffile(match.group(1), match.group(2))
+ elif status == "pmstatus":
+ # FIXME: Float comparison
+ if float(percent) != self.percent or status_str != self.status:
+ self.status_change(pkgname, float(percent), status_str.strip())
+ self.percent = float(percent)
+ self.status = status_str.strip()
+ elif base == "status":
+ self.dpkg_status_change(pkgname, status)
+
+ def wait_child(self):
+ """Wait for child progress to exit.
+
+ This method is responsible for calling update_interface() from time to
+ time. It exits once the child has exited. The return values is the
+ full status returned from os.waitpid() (not only the return code).
+ """
+ (pid, res) = (0, 0)
+ while True:
+ try:
+ select.select([self.statusfd], [], [], self.select_timeout)
+ except select.error, (errno_, errstr):
+ if errno_ != errno.EINTR:
+ raise
+
+ self.update_interface()
+ try:
+ (pid, res) = os.waitpid(self.child_pid, os.WNOHANG)
+ if pid == self.child_pid:
+ break
+ except OSError, err:
+ if err.errno == errno.ECHILD:
+ break
+ if err.errno != errno.EINTR:
+ raise
+
+ return res
+
+
+class OpProgress(object):
+ """Monitor objects for operations.
+
+ Display the progress of operations such as opening the cache."""
+
+ major_change, op, percent, subop = False, "", 0.0, ""
+
+ def update(self, percent=None):
+ """Called periodically to update the user interface.
+
+ You may use the optional argument 'percent' to set the attribute
+ 'percent' in this call.
+ """
+ if percent is not None:
+ self.percent = percent
+
+ def done(self):
+ """Called once an operation has been completed."""
diff --git a/apt/progress/gtk2.py b/apt/progress/gtk2.py
index f872e34f..29e730a3 100644
--- a/apt/progress/gtk2.py
+++ b/apt/progress/gtk2.py
@@ -11,7 +11,7 @@
# published by the Free Software Foundation; either version 2 of the
# License, or (at your option) any later version.
#
-# his program is distributed in the hope that it will be useful,
+# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
@@ -22,7 +22,6 @@
# USA
"""GObject-powered progress classes and a GTK+ status widget."""
-from gettext import gettext as _
import os
import time
@@ -37,8 +36,17 @@ import gobject
import pango
import vte
-import apt
import apt_pkg
+from apt_pkg import gettext as _
+from apt.deprecation import function_deprecated_by, AttributeDeprecatedBy
+from apt.progress import base
+
+if apt_pkg._COMPAT_0_7:
+ from apt.progress import old
+
+
+__all__ = ['GAcquireProgress', 'GInstallProgress', 'GOpProgress',
+ 'GtkAptProgress']
def mksig(params=(), run=gobject.SIGNAL_RUN_FIRST, rettype=gobject.TYPE_NONE):
@@ -52,7 +60,7 @@ def mksig(params=(), run=gobject.SIGNAL_RUN_FIRST, rettype=gobject.TYPE_NONE):
return (run, rettype, params)
-class GOpProgress(gobject.GObject, apt.progress.OpProgress):
+class GOpProgress(gobject.GObject, base.OpProgress):
"""Operation progress with GObject signals.
Signals:
@@ -68,22 +76,28 @@ class GOpProgress(gobject.GObject, apt.progress.OpProgress):
"status-finished": mksig()}
def __init__(self):
- apt.progress.OpProgress.__init__(self)
+ base.OpProgress.__init__(self)
gobject.GObject.__init__(self)
self._context = glib.main_context_default()
- def update(self, percent):
+ def update(self, percent=None):
"""Called to update the percentage done"""
- self.emit("status-changed", self.op, percent)
+ base.OpProgress.update(self, percent)
+ self.emit("status-changed", self.op, self.percent)
while self._context.pending():
self._context.iteration()
def done(self):
"""Called when all operation have finished."""
+ base.OpProgress.done(self)
self.emit("status-finished")
+ if apt_pkg._COMPAT_0_7:
+ subOp = AttributeDeprecatedBy('subop')
+ Op = AttributeDeprecatedBy('op')
-class GInstallProgress(gobject.GObject, apt.progress.InstallProgress):
+
+class GInstallProgress(gobject.GObject, base.InstallProgress):
"""Installation progress with GObject signals.
Signals:
@@ -107,19 +121,19 @@ class GInstallProgress(gobject.GObject, apt.progress.InstallProgress):
"status-finished": mksig()}
def __init__(self, term):
- apt.progress.InstallProgress.__init__(self)
+ base.InstallProgress.__init__(self)
gobject.GObject.__init__(self)
self.finished = False
self.time_last_update = time.time()
self.term = term
reaper = vte.reaper_get()
- reaper.connect("child-exited", self.childExited)
+ reaper.connect("child-exited", self.child_exited)
self.env = ["VTE_PTY_KEEP_FD=%s" % self.writefd,
"DEBIAN_FRONTEND=gnome",
"APT_LISTCHANGES_FRONTEND=gtk"]
self._context = glib.main_context_default()
- def childExited(self, term, pid, status):
+ def child_exited(self, term, pid, status):
"""Called when a child process exits"""
self.apt_status = os.WEXITSTATUS(status)
self.finished = True
@@ -138,21 +152,31 @@ class GInstallProgress(gobject.GObject, apt.progress.InstallProgress):
"""
self.emit("status-conffile")
- def startUpdate(self):
+ def start_update(self):
"""Called when the update starts.
Emits: status-started()
"""
self.emit("status-started")
- def finishUpdate(self):
+ def run(self, obj):
+ """Run."""
+ self.finished = False
+ return base.InstallProgress.run(self, obj)
+
+ def finish_update(self):
"""Called when the update finished.
Emits: status-finished()
"""
self.emit("status-finished")
- def statusChange(self, pkg, percent, status):
+ def processing(self, pkg, stage):
+ """Called when entering a new stage in dpkg."""
+ # We have no percentage or alike, send -1 to let the bar pulse.
+ self.emit("status-changed", ("Installing %s...") % pkg, -1)
+
+ def status_change(self, pkg, percent, status):
"""Called when the status changed.
Emits: status-changed(status, percent)
@@ -160,12 +184,12 @@ class GInstallProgress(gobject.GObject, apt.progress.InstallProgress):
self.time_last_update = time.time()
self.emit("status-changed", status, percent)
- def updateInterface(self):
+ def update_interface(self):
"""Called periodically to update the interface.
Emits: status-timeout() [When a timeout happens]
"""
- apt.progress.InstallProgress.updateInterface(self)
+ base.InstallProgress.update_interface(self)
while self._context.pending():
self._context.iteration()
if self.time_last_update + self.INSTALL_TIMEOUT < time.time():
@@ -175,40 +199,25 @@ class GInstallProgress(gobject.GObject, apt.progress.InstallProgress):
"""Fork the process."""
return self.term.forkpty(envv=self.env)
- def waitChild(self):
+ def wait_child(self):
"""Wait for the child process to exit."""
while not self.finished:
- self.updateInterface()
+ self.update_interface()
return self.apt_status
+ if apt_pkg._COMPAT_0_7:
+ updateInterface = function_deprecated_by(update_interface)
+ startUpdate = function_deprecated_by(start_update)
+ finishUpdate = function_deprecated_by(finish_update)
+ statusChange = function_deprecated_by(status_change)
+ waitChild = function_deprecated_by(wait_child)
+ childExited = function_deprecated_by(child_exited)
-class GDpkgInstallProgress(apt.progress.DpkgInstallProgress, GInstallProgress):
- """An InstallProgress for local installations.
-
- Signals:
-
- * status-changed(str: status, int: percent)
- * status-started() - Not Implemented yet
- * status-finished()
- * status-timeout() - When the maintainer script hangs
- * status-error() - When an error happens
- * status-conffile() - On Conffile
- """
-
- def run(self, debfile):
- """Install the given package."""
- apt.progress.DpkgInstallProgress.run(self, debfile)
-
- def updateInterface(self):
- """Called periodically to update the interface.
- Emits: status-timeout() [When a timeout happens]"""
- apt.progress.DpkgInstallProgress.updateInterface(self)
- if self.time_last_update + self.INSTALL_TIMEOUT < time.time():
- self.emit("status-timeout")
+GDpkgInstallProgress = GInstallProgress
-class GFetchProgress(gobject.GObject, apt.progress.FetchProgress):
+class GAcquireProgress(gobject.GObject, base.AcquireProgress):
"""A Fetch Progress with GObject signals.
Signals:
@@ -216,6 +225,8 @@ class GFetchProgress(gobject.GObject, apt.progress.FetchProgress):
* status-changed(str: description, int: percent)
* status-started()
* status-finished()
+
+ DEPRECATED.
"""
__gsignals__ = {"status-changed": mksig((str, int)),
@@ -223,40 +234,98 @@ class GFetchProgress(gobject.GObject, apt.progress.FetchProgress):
"status-finished": mksig()}
def __init__(self):
- apt.progress.FetchProgress.__init__(self)
+ base.AcquireProgress.__init__(self)
gobject.GObject.__init__(self)
self._continue = True
self._context = glib.main_context_default()
def start(self):
+ base.AcquireProgress.start(self)
self.emit("status-started")
def stop(self):
+ base.AcquireProgress.stop(self)
self.emit("status-finished")
def cancel(self):
self._continue = False
- def pulse(self):
- apt.progress.FetchProgress.pulse(self)
- currentItem = self.currentItems + 1
- if currentItem > self.totalItems:
- currentItem = self.totalItems
- if self.currentCPS > 0:
+ def pulse(self, owner):
+ base.AcquireProgress.pulse(self, owner)
+ current_item = self.current_items + 1
+ if current_item > self.total_items:
+ current_item = self.total_items
+ if self.current_cps > 0:
text = (_("Downloading file %(current)li of %(total)li with "
"%(speed)s/s") % \
- {"current": currentItem,
- "total": self.totalItems,
- "speed": apt_pkg.SizeToStr(self.currentCPS)})
+ {"current": current_item,
+ "total": self.total_items,
+ "speed": apt_pkg.size_to_str(self.current_cps)})
else:
text = (_("Downloading file %(current)li of %(total)li") % \
- {"current": currentItem,
- "total": self.totalItems})
- self.emit("status-changed", text, self.percent)
+ {"current": current_item,
+ "total": self.total_items})
+
+ percent = (((self.current_bytes + self.current_items) * 100.0) /
+ float(self.total_bytes + self.total_items))
+ self.emit("status-changed", text, percent)
while self._context.pending():
self._context.iteration()
return self._continue
+if apt_pkg._COMPAT_0_7:
+
+ class GFetchProgress(gobject.GObject, old.FetchProgress):
+ """A Fetch Progress with GObject signals.
+
+ Signals:
+
+ * status-changed(str: description, int: percent)
+ * status-started()
+ * status-finished()
+
+ DEPRECATED.
+ """
+
+ __gsignals__ = {"status-changed": mksig((str, int)),
+ "status-started": mksig(),
+ "status-finished": mksig()}
+
+ def __init__(self):
+ old.FetchProgress.__init__(self)
+ gobject.GObject.__init__(self)
+ self._continue = True
+ self._context = glib.main_context_default()
+
+ def start(self):
+ self.emit("status-started")
+
+ def stop(self):
+ self.emit("status-finished")
+
+ def cancel(self):
+ self._continue = False
+
+ def pulse(self):
+ old.FetchProgress.pulse(self)
+ current_item = self.currentItems + 1
+ if current_item > self.totalItems:
+ current_item = self.totalItems
+ if self.current_cps > 0:
+ text = (_("Downloading file %(current)li of %(total)li with "
+ "%(speed)s/s") % \
+ {"current": current_item,
+ "total": self.totalItems,
+ "speed": apt_pkg.size_to_str(self.currentCPS)})
+ else:
+ text = (_("Downloading file %(current)li of %(total)li") % \
+ {"current": current_item,
+ "total": self.totalItems})
+ self.emit("status-changed", text, self.percent)
+ while self._context.pending():
+ self._context.iteration()
+ return self._continue
+
class GtkAptProgress(gtk.VBox):
"""Graphical progress for installation/fetch/operations.
@@ -291,11 +360,15 @@ class GtkAptProgress(gtk.VBox):
self._progress_open.connect("status-started", self._on_status_started)
self._progress_open.connect("status-finished",
self._on_status_finished)
- self._progress_fetch = GFetchProgress()
- self._progress_fetch.connect("status-changed", self._on_status_changed)
- self._progress_fetch.connect("status-started", self._on_status_started)
- self._progress_fetch.connect("status-finished",
+ self._progress_acquire = GAcquireProgress()
+ self._progress_acquire.connect("status-changed",
+ self._on_status_changed)
+ self._progress_acquire.connect("status-started",
+ self._on_status_started)
+ self._progress_acquire.connect("status-finished",
self._on_status_finished)
+
+ self._progress_fetch = None
self._progress_install = GInstallProgress(self._terminal)
self._progress_install.connect("status-changed",
self._on_status_changed)
@@ -309,19 +382,6 @@ class GtkAptProgress(gtk.VBox):
self._on_status_timeout)
self._progress_install.connect("status-conffile",
self._on_status_timeout)
- self._progress_dpkg_install = GDpkgInstallProgress(self._terminal)
- self._progress_dpkg_install.connect("status-changed",
- self._on_status_changed)
- self._progress_dpkg_install.connect("status-started",
- self._on_status_started)
- self._progress_dpkg_install.connect("status-finished",
- self._on_status_finished)
- self._progress_dpkg_install.connect("status-timeout",
- self._on_status_timeout)
- self._progress_dpkg_install.connect("status-error",
- self._on_status_timeout)
- self._progress_dpkg_install.connect("status-conffile",
- self._on_status_timeout)
def clear(self):
"""Reset all status information."""
@@ -342,12 +402,27 @@ class GtkAptProgress(gtk.VBox):
@property
def dpkg_install(self):
"""Return the install progress handler for dpkg."""
- return self._dpkg_progress_install
+ return self._progress_install
+
+ if apt_pkg._COMPAT_0_7:
+
+ @property
+ def fetch(self):
+ """Return the fetch progress handler."""
+ if self._progress_fetch is None:
+ self._progress_fetch = GFetchProgress()
+ self._progress_fetch.connect("status-changed",
+ self._on_status_changed)
+ self._progress_fetch.connect("status-started",
+ self._on_status_started)
+ self._progress_fetch.connect("status-finished",
+ self._on_status_finished)
+ return self._progress_fetch
@property
- def fetch(self):
- """Return the fetch progress handler."""
- return self._progress_fetch
+ def acquire(self):
+ """Return the acquire progress handler."""
+ return self._progress_acquire
def _on_status_started(self, progress):
"""Called when something starts."""
@@ -364,7 +439,7 @@ class GtkAptProgress(gtk.VBox):
def _on_status_changed(self, progress, status, percent):
"""Called when the status changed."""
self._label.set_text(status)
- if percent is None:
+ if percent is None or percent == -1:
self._progressbar.pulse()
else:
self._progressbar.set_fraction(percent/100.0)
@@ -412,6 +487,7 @@ def _test():
"""Test function"""
import sys
+ import apt
from apt.debfile import DebPackage
win = gtk.Window()
@@ -422,18 +498,19 @@ def _test():
win.show()
cache = apt.cache.Cache(apt_progress.open)
pkg = cache["xterm"]
- if pkg.isInstalled:
- pkg.markDelete()
+ if pkg.is_installed:
+ pkg.mark_delete()
else:
- pkg.markInstall()
+ pkg.mark_install()
apt_progress.show_terminal(True)
try:
- cache.commit(apt_progress.fetch, apt_progress.install)
+ cache.commit(apt_progress.acquire, apt_progress.install)
except Exception, exc:
print >> sys.stderr, "Exception happened:", exc
if len(sys.argv) > 1:
deb = DebPackage(sys.argv[1], cache)
deb.install(apt_progress.dpkg_install)
+ win.connect("destroy", gtk.main_quit)
gtk.main()
diff --git a/apt/progress/old.py b/apt/progress/old.py
new file mode 100644
index 00000000..c2d95b85
--- /dev/null
+++ b/apt/progress/old.py
@@ -0,0 +1,205 @@
+# progress.py - progress reporting classes
+#
+# Copyright (c) 2005-2009 Canonical
+#
+# Author: Michael Vogt <michael.vogt@ubuntu.com>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+"""Deprecated progress reporting classes.
+
+This module provides classes for compatibility with python-apt 0.7. They are
+completely deprecated and should not be used anymore.
+"""
+
+
+import os
+import sys
+
+import apt_pkg
+from apt.deprecation import AttributeDeprecatedBy, function_deprecated_by
+import warnings
+from apt.progress import base, text
+
+__all__ = []
+
+
+class OpProgress(base.OpProgress):
+ """Abstract class to implement reporting on cache opening."""
+
+ subOp = AttributeDeprecatedBy('subop')
+ Op = AttributeDeprecatedBy('op')
+
+
+class OpTextProgress(OpProgress, text.OpProgress):
+ """A simple text based cache open reporting class."""
+
+
+class FetchProgress(object):
+ """Report the download/fetching progress."""
+
+ # download status constants
+ (dlDone, dlQueued, dlFailed, dlHit, dlIgnored) = range(5)
+ dlStatusStr = {dlDone: "Done", dlQueued: "Queued", dlFailed: "Failed",
+ dlHit: "Hit", dlIgnored: "Ignored"}
+
+ def __init__(self):
+ self.eta = 0.0
+ self.percent = 0.0
+ # Make checking easier
+ self.currentBytes = 0
+ self.currentItems = 0
+ self.totalBytes = 0
+ self.totalItems = 0
+ self.currentCPS = 0
+ warnings.warn("FetchProgress() is deprecated.", DeprecationWarning)
+
+ def start(self):
+ """Called when the fetching starts."""
+
+ def stop(self):
+ """Called when all files have been fetched."""
+
+ def updateStatus(self, uri, descr, short_descr, status):
+ """Called when the status of an item changes.
+
+ This happens eg. when the downloads fails or is completed.
+ """
+
+ def update_status_full(self, uri, descr, short_descr, status, file_size,
+ partial_size):
+ """Called when the status of an item changes.
+
+ This happens eg. when the downloads fails or is completed. This
+ version include information on current filesize and partial size
+ """
+
+ def pulse(self):
+ """Called periodically to update the user interface.
+
+ Return True to continue or False to cancel.
+ """
+ self.percent = (((self.currentBytes + self.currentItems) * 100.0) /
+ float(self.totalBytes + self.totalItems))
+ if self.currentCPS > 0:
+ self.eta = ((self.totalBytes - self.currentBytes) /
+ float(self.currentCPS))
+ return True
+
+ def pulse_items(self, items):
+ """Called periodically to update the user interface.
+ This function includes details about the items being fetched
+ Return True to continue or False to cancel.
+
+ """
+ self.percent = (((self.currentBytes + self.currentItems) * 100.0) /
+ float(self.totalBytes + self.totalItems))
+ if self.currentCPS > 0:
+ self.eta = ((self.totalBytes - self.currentBytes) /
+ float(self.currentCPS))
+ return True
+
+ def mediaChange(self, medium, drive):
+ """react to media change events."""
+
+
+class TextFetchProgress(FetchProgress):
+ """ Ready to use progress object for terminal windows """
+
+ def __init__(self):
+ FetchProgress.__init__(self)
+ self.items = {}
+
+ def updateStatus(self, uri, descr, short_descr, status):
+ """Called when the status of an item changes.
+
+ This happens eg. when the downloads fails or is completed.
+ """
+ if status != self.dlQueued:
+ print "\r%s %s" % (self.dlStatusStr[status], descr)
+ self.items[uri] = status
+
+ def pulse(self):
+ """Called periodically to update the user interface.
+
+ Return True to continue or False to cancel.
+ """
+ FetchProgress.pulse(self)
+ if self.currentCPS > 0:
+ s = "[%2.f%%] %sB/s %s" % (self.percent,
+ apt_pkg.size_to_str(int(self.currentCPS)),
+ apt_pkg.time_to_str(int(self.eta)))
+ else:
+ s = "%2.f%% [Working]" % (self.percent)
+ print "\r%s" % (s),
+ sys.stdout.flush()
+ return True
+
+ def stop(self):
+ """Called when all files have been fetched."""
+ print "\rDone downloading "
+
+ def mediaChange(self, medium, drive):
+ """react to media change events."""
+ print ("Media change: please insert the disc labeled "
+ "'%s' in the drive '%s' and press enter") % (medium, drive)
+
+ return raw_input() not in ('c', 'C')
+
+
+class CdromProgress(base.CdromProgress):
+ """Report the cdrom add progress.
+
+ This class has been replaced by apt_pkg.CdromProgress.
+ """
+ _basetype = base.CdromProgress
+ askCdromName = function_deprecated_by(_basetype.ask_cdrom_name)
+ changeCdrom = function_deprecated_by(_basetype.change_cdrom)
+ del _basetype
+
+
+class DumbInstallProgress(base.InstallProgress):
+ """Report the install progress.
+
+ Subclass this class to implement install progress reporting.
+ """
+
+ startUpdate = function_deprecated_by(base.InstallProgress.start_update)
+ finishUpdate = function_deprecated_by(base.InstallProgress.finish_update)
+ updateInterface = function_deprecated_by(
+ base.InstallProgress.update_interface)
+
+
+class InstallProgress(DumbInstallProgress, base.InstallProgress):
+ """An InstallProgress that is pretty useful.
+
+ It supports the attributes 'percent' 'status' and callbacks for the dpkg
+ errors and conffiles and status changes.
+ """
+
+ selectTimeout = AttributeDeprecatedBy('select_timeout')
+ statusChange = function_deprecated_by(base.InstallProgress.status_change)
+ waitChild = function_deprecated_by(base.InstallProgress.wait_child)
+
+
+class DpkgInstallProgress(InstallProgress):
+ """Progress handler for a local Debian package installation."""
+
+ def run(self, debfile):
+ """Start installing the given Debian package."""
+ # Deprecated stuff
+ self.debfile = debfile
+ self.debname = os.path.basename(debfile).split("_")[0]
+ return base.InstallProgress(self, debfile)
diff --git a/apt/progress/text.py b/apt/progress/text.py
new file mode 100644
index 00000000..5e45c1db
--- /dev/null
+++ b/apt/progress/text.py
@@ -0,0 +1,261 @@
+# Copyright (c) 2009 Julian Andres Klode <jak@debian.org>
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+# USA
+"""Progress reporting for text interfaces."""
+import sys
+
+import apt_pkg
+from apt.progress import base
+
+__all__ = ['AcquireProgress', 'CdromProgress', 'OpProgress']
+
+
+def _(msg):
+ """Translate the message, also try apt if translation is missing."""
+ res = apt_pkg.gettext(msg)
+ if res == msg:
+ res = apt_pkg.gettext(msg, "apt")
+ return res
+
+
+class TextProgress(object):
+ """Internal Base class for text progress classes."""
+
+ def __init__(self, outfile=None):
+ self._file = outfile or sys.stdout
+ self._width = 0
+
+ def _write(self, msg, newline=True, maximize=False):
+ """Write the message on the terminal, fill remaining space."""
+ self._file.write("\r")
+ self._file.write(msg)
+
+ # Fill remaining stuff with whitespace
+ if self._width > len(msg):
+ self._file.write((self._width - len(msg)) * ' ')
+ elif maximize: # Needed for OpProgress.
+ self._width = max(self._width, len(msg))
+ if newline:
+ self._file.write("\n")
+ else:
+ #self._file.write("\r")
+ self._file.flush()
+
+
+class OpProgress(base.OpProgress, TextProgress):
+ """Operation progress reporting.
+
+ This closely resembles OpTextProgress in libapt-pkg.
+ """
+
+ def __init__(self, outfile=None):
+ TextProgress.__init__(self, outfile)
+ base.OpProgress.__init__(self)
+ self.old_op = ""
+
+ def update(self, percent=None):
+ """Called periodically to update the user interface."""
+ base.OpProgress.update(self, percent)
+ if self.major_change and self.old_op:
+ self._write(self.old_op)
+ self._write("%s... %i%%\r" % (self.op, self.percent), False, True)
+ self.old_op = self.op
+
+ def done(self):
+ """Called once an operation has been completed."""
+ base.OpProgress.done(self)
+ if self.old_op:
+ self._write(_("%c%s... Done") % ('\r', self.old_op), True, True)
+ self.old_op = ""
+
+
+class AcquireProgress(base.AcquireProgress, TextProgress):
+ """AcquireProgress for the text interface."""
+
+ def __init__(self, outfile=None):
+ TextProgress.__init__(self, outfile)
+ base.AcquireProgress.__init__(self)
+ self._signal = None
+ self._width = 80
+ self._id = 1
+
+ def start(self):
+ """Start an Acquire progress.
+
+ In this case, the function sets up a signal handler for SIGWINCH, i.e.
+ window resize signals. And it also sets id to 1.
+ """
+ base.AcquireProgress.start(self)
+ import signal
+ self._signal = signal.signal(signal.SIGWINCH, self._winch)
+ # Get the window size.
+ self._winch()
+ self._id = 1L
+
+ def _winch(self, *dummy):
+ """Signal handler for window resize signals."""
+ import fcntl
+ import termios
+ import struct
+ buf = fcntl.ioctl(self._file, termios.TIOCGWINSZ, 8 * ' ')
+ dummy, col, dummy, dummy = struct.unpack('hhhh', buf)
+ self._width = col - 1 # 1 for the cursor
+
+ def ims_hit(self, item):
+ """Called when an item is update (e.g. not modified on the server)."""
+ base.AcquireProgress.ims_hit(self, item)
+ line = _('Hit ') + item.description
+ if item.owner.filesize:
+ line += ' [%sB]' % apt_pkg.size_to_str(item.owner.filesize)
+ self._write(line)
+
+ def fail(self, item):
+ """Called when an item is failed."""
+ base.AcquireProgress.fail(self, item)
+ if item.owner.status == item.owner.STAT_DONE:
+ self._write(_("Ign ") + item.description)
+ else:
+ self._write(_("Err ") + item.description)
+ self._write(" %s" % item.owner.error_text)
+
+ def fetch(self, item):
+ """Called when some of the item's data is fetched."""
+ base.AcquireProgress.fetch(self, item)
+ # It's complete already (e.g. Hit)
+ if item.owner.complete:
+ return
+ item.owner.id = self._id
+ self._id += 1
+ line = _("Get:") + "%s %s" % (item.owner.id, item.description)
+ if item.owner.filesize:
+ line += (" [%sB]" % apt_pkg.size_to_str(item.owner.filesize))
+
+ self._write(line)
+
+ def pulse(self, owner):
+ """Periodically invoked while the Acquire process is underway.
+
+ Return False if the user asked to cancel the whole Acquire process."""
+ base.AcquireProgress.pulse(self, owner)
+ percent = (((self.current_bytes + self.current_items) * 100.0) /
+ float(self.total_bytes + self.total_items))
+
+ shown = False
+ tval = '%i%%' % percent
+
+ end = ""
+ if self.current_cps:
+ eta = int(float(self.total_bytes - self.current_bytes) /
+ self.current_cps)
+ end = " %sB/s %s" % (apt_pkg.size_to_str(self.current_cps),
+ apt_pkg.time_to_str(eta))
+
+ for worker in owner.workers:
+ val = ''
+ if not worker.current_item:
+ if worker.status:
+ val = ' [%s]' % worker.status
+ if len(tval) + len(val) + len(end) >= self._width:
+ break
+ tval += val
+ shown = True
+ continue
+ shown = True
+
+ if worker.current_item.owner.id:
+ val += " [%i %s" % (worker.current_item.owner.id,
+ worker.current_item.shortdesc)
+ else:
+ val += ' [%s' % worker.current_item.description
+ if worker.current_item.owner.mode:
+ val += ' %s' % worker.current_item.owner.mode
+
+ val += ' %sB' % apt_pkg.size_to_str(worker.current_size)
+
+ # Add the total size and percent
+ if worker.total_size and not worker.current_item.owner.complete:
+ val += "/%sB %i%%" % (apt_pkg.size_to_str(worker.total_size),
+ worker.current_size*100.0/worker.total_size)
+
+ val += ']'
+
+ if len(tval) + len(val) + len(end) >= self._width:
+ # Display as many items as screen width
+ break
+ else:
+ tval += val
+
+ if not shown:
+ tval += _(" [Working]")
+
+ if self.current_cps:
+ tval += (self._width - len(end) - len(tval)) * ' ' + end
+
+ self._write(tval, False)
+ return True
+
+ def media_change(self, medium, drive):
+ """Prompt the user to change the inserted removable media."""
+ base.AcquireProgress.media_change(self, medium, drive)
+ self._write(_("Media change: please insert the disc labeled\n"
+ " '%s'\n"
+ "in the drive '%s' and press enter\n") % (medium, drive))
+ return raw_input() not in ('c', 'C')
+
+ def stop(self):
+ """Invoked when the Acquire process stops running."""
+ base.AcquireProgress.stop(self)
+ # Trick for getting a translation from apt
+ self._write((_("Fetched %sB in %s (%sB/s)\n") % (
+ apt_pkg.size_to_str(self.fetched_bytes),
+ apt_pkg.time_to_str(self.elapsed_time),
+ apt_pkg.size_to_str(self.current_cps))).rstrip("\n"))
+
+ # Delete the signal again.
+ import signal
+ signal.signal(signal.SIGWINCH, self._signal)
+
+
+class CdromProgress(base.CdromProgress, TextProgress):
+ """Text CD-ROM progress."""
+
+ def ask_cdrom_name(self):
+ """Ask the user to provide a name for the disc."""
+ base.CdromProgress.ask_cdrom_name(self)
+ self._write(_("Please provide a name for this Disc, such as "
+ "'Debian 2.1r1 Disk 1'"), False)
+ try:
+ return raw_input(":")
+ except KeyboardInterrupt:
+ return
+
+ def update(self, text, current):
+ """Set the current progress."""
+ base.CdromProgress.update(self, text, current)
+ if text:
+ self._write(text, False)
+
+ def change_cdrom(self):
+ """Ask the user to change the CD-ROM."""
+ base.CdromProgress.change_cdrom(self)
+ self._write(_("Please insert a Disc in the drive and press enter"),
+ False)
+ try:
+ return (raw_input() == '')
+ except KeyboardInterrupt:
+ return False
+
+InstallProgress = base.InstallProgress
diff --git a/apt/utils.py b/apt/utils.py
index df1d0397..61d5d54f 100644
--- a/apt/utils.py
+++ b/apt/utils.py
@@ -38,39 +38,39 @@ def get_release_date_from_release_file(path):
"""
if not path or not os.path.exists(path):
return None
- tag = apt_pkg.ParseTagFile(open(path))
- tag.Step()
- if not tag.Section.has_key("Date"):
+ tag = apt_pkg.TagFile(open(path))
+ section = tag.next()
+ if not "Date" in section:
return None
- date = tag.Section["Date"]
- return apt_pkg.StrToTime(date)
+ date = section["Date"]
+ return apt_pkg.str_to_time(date)
def get_release_filename_for_pkg(cache, pkgname, label, release):
" get the release file that provides this pkg "
- if not cache.has_key(pkgname):
+ if pkgname not in cache:
return None
pkg = cache[pkgname]
ver = None
# look for the version that comes from the repos with
# the given label and origin
- for aver in pkg._pkg.VersionList:
- if aver == None or aver.FileList == None:
+ for aver in pkg._pkg.version_list:
+ if aver == None or aver.file_list == None:
continue
- for verFile, index in aver.FileList:
+ for ver_file, index in aver.file_list:
#print verFile
- if (verFile.Origin == label and
- verFile.Label == label and
- verFile.Archive == release):
+ if (ver_file.origin == label and
+ ver_file.label == label and
+ ver_file.archive == release):
ver = aver
if not ver:
return None
- indexfile = cache._list.FindIndex(ver.FileList[0][0])
- for metaindex in cache._list.List:
- for m in metaindex.IndexFiles:
+ indexfile = cache._list.find_index(ver.file_list[0][0])
+ for metaindex in cache._list.list:
+ for m in metaindex.index_files:
if (indexfile and
- indexfile.Describe == m.Describe and
- indexfile.IsTrusted):
- dir = apt_pkg.Config.FindDir("Dir::State::lists")
- name = apt_pkg.URItoFileName(metaindex.URI)+"dists_%s_Release" % metaindex.Dist
+ indexfile.describe == m.describe and
+ indexfile.is_trusted):
+ dir = apt_pkg.config.find_dir("Dir::State::lists")
+ name = apt_pkg.uri_to_filename(metaindex.uri)+"dists_%s_Release" % metaindex.dist
return dir+name
return None
generated by cgit v1.2.3 (git 2.46.0) at 2026-05-10 20:53:25 +0000