pull-pkg: change pullpkg into class PullPkg

instead of pullpkg.py containing a simple method to call, change it
into a normal class PullPkg that callers can create and use.
This commit is contained in:
Dan Streetman 2018-07-10 14:20:44 -04:00
parent fb750e38bb
commit 3491b0cff9
2 changed files with 266 additions and 212 deletions

View File

@ -128,8 +128,7 @@ class SourcePackage(object):
spph_class = SourcePackagePublishingHistory spph_class = SourcePackagePublishingHistory
def __init__(self, package=None, version=None, component=None, def __init__(self, package=None, version=None, component=None,
dscfile=None, lp=None, mirrors=(), workdir='.', quiet=False, *args, **kwargs):
series=None, pocket=None, verify_signature=False):
"""Can be initialised using either package or dscfile. """Can be initialised using either package or dscfile.
If package is specified, either the version or series can also be If package is specified, either the version or series can also be
specified; using version will get the specific package version, specified; using version will get the specific package version,
@ -137,6 +136,15 @@ class SourcePackage(object):
Specifying only the package with no version or series will get the Specifying only the package with no version or series will get the
latest version from the development series. latest version from the development series.
""" """
dscfile = kwargs.get('dscfile')
lp = kwargs.get('lp')
mirrors = kwargs.get('mirrors', ())
workdir = kwargs.get('workdir', '.')
quiet = kwargs.get('quiet', False)
series = kwargs.get('series')
pocket = kwargs.get('pocket')
verify_signature = kwargs.get('verify_signature', False)
assert (package is not None or dscfile is not None) assert (package is not None or dscfile is not None)
self.source = package self.source = package

View File

@ -23,6 +23,8 @@
import re import re
import errno
from argparse import ArgumentParser from argparse import ArgumentParser
from distro_info import DebianDistroInfo from distro_info import DebianDistroInfo
@ -63,221 +65,265 @@ class InvalidPullValueError(ValueError):
pass pass
def create_argparser(default_pull=None, default_distro=None, default_arch=None): class PullPkg(object):
help_default_pull = "What to pull: " + ", ".join(VALID_PULLS) """Class used to pull file(s) associated with a specific package"""
if default_pull: @classmethod
help_default_pull += (" (default: %s)" % default_pull) def main(cls, *args, **kwargs):
help_default_distro = "Pull from: " + ", ".join(VALID_DISTROS) """For use by stand-alone cmdline scripts.
if default_distro:
help_default_distro += (" (default: %s)" % default_distro)
if not default_arch:
default_arch = host_architecture()
help_default_arch = ("Get binary packages for arch (default: %s)" % default_arch)
parser = ArgumentParser() This will handle catching certain exceptions or kbd interrupts,
parser.add_argument('-v', '--verbose', action='store_true', and printing out (via Logger) the error message, instead of
help="Print verbose/debug messages") allowing the exception to flow up to the script. This does
parser.add_argument('-d', '--download-only', action='store_true', not catch unexpected exceptions, such as internal errors.
help="Do not extract the source package") On (expected) error, this will call sys.exit(error);
parser.add_argument('-m', '--mirror', action='append', unexpected errors will flow up to the caller.
help='Preferred mirror(s)') On success, this simply returns.
parser.add_argument('--no-conf', action='store_true', """
help="Don't read config files or environment variables")
parser.add_argument('--no-verify-signature', action='store_true',
help="Don't fail if dsc signature can't be verified")
parser.add_argument('-a', '--arch', default=default_arch,
help=help_default_arch)
parser.add_argument('-p', '--pull', default=default_pull,
help=help_default_pull)
parser.add_argument('-D', '--distro', default=default_distro,
help=help_default_distro)
parser.add_argument('package', help="Package name to pull")
parser.add_argument('release', nargs='?', help="Release to pull from")
parser.add_argument('version', nargs='?', help="Package version to pull")
return parser
def parse_pull(pull):
if not pull:
raise InvalidPullValueError("Must specify --pull")
# allow 'dbgsym' as alias for 'ddebs'
if pull == 'dbgsym':
Logger.debug("Pulling '%s' for '%s'", PULL_DDEBS, pull)
pull = PULL_DDEBS
# assume anything starting with 'bin' means 'debs'
if str(pull).startswith('bin'):
Logger.debug("Pulling '%s' for '%s'", PULL_DEBS, pull)
pull = PULL_DEBS
# verify pull action is valid
if pull not in VALID_PULLS:
raise InvalidPullValueError("Invalid pull action '%s'" % pull)
return pull
def parse_distro(distro):
if not distro:
raise InvalidDistroValueError("Must specify --distro")
distro = distro.lower()
# allow 'lp' for 'ubuntu'
if distro == 'lp':
Logger.debug("Using distro '%s' for '%s'", DISTRO_UBUNTU, distro)
distro = DISTRO_UBUNTU
# assume anything with 'cloud' is UCA
if re.match(r'.*cloud.*', distro):
Logger.debug("Using distro '%s' for '%s'", DISTRO_UCA, distro)
distro = DISTRO_UCA
# verify distro is valid
if distro not in VALID_DISTROS:
raise InvalidDistroValueError("Invalid distro '%s'" % distro)
return distro
def parse_release(release, distro):
if distro == DISTRO_UCA:
# UCA is special; it is specified UBUNTURELEASE-UCARELEASE or just
# UCARELEASE. The user could also specify UCARELEASE-POCKET. But UCA
# archives always correspond to only one UBUNTURELEASE, and UCA archives
# have only the Release pocket, so only UCARELEASE matters to us.
for r in release.split('-'):
if r in UbuntuCloudArchiveSourcePackage.getReleaseNames():
Logger.debug("Using UCA release '%s'", r)
return (r, None)
raise SeriesNotFoundException('UCA release {} not found.'.format(release))
# Check if release[-pocket] is specified
(release, pocket) = split_release_pocket(release, default=None)
Logger.debug("Parsed release '%s' pocket '%s'", release, pocket)
if distro == DISTRO_DEBIAN:
# This converts from the aliases like 'unstable'
debian_info = DebianDistroInfo()
codename = debian_info.codename(release)
if codename:
Logger.normal("Using release '%s' for '%s'", codename, release)
release = codename
d = Distribution(distro)
# let SeriesNotFoundException flow up
d.getSeries(release)
Logger.debug("Using distro '%s' release '%s' pocket '%s'",
distro, release, pocket)
return (release, pocket)
def pull(options):
# required options asserted below
# 'release' and 'version' are optional strings
# 'mirror' is optional list of strings
# these are type bool
assert hasattr(options, 'verbose')
assert hasattr(options, 'download_only')
assert hasattr(options, 'no_conf')
assert hasattr(options, 'no_verify_signature')
# these are type string
assert hasattr(options, 'arch')
assert hasattr(options, 'pull')
assert hasattr(options, 'distro')
assert hasattr(options, 'package')
Logger.set_verbosity(options.verbose)
Logger.debug("pullpkg options: %s", options)
# Login anonymously to LP
Launchpad.login_anonymously()
pull = parse_pull(options.pull)
distro = parse_distro(options.distro)
config = UDTConfig(options.no_conf)
mirrors = []
if hasattr(options, 'mirror') and options.mirror:
mirrors += options.mirror
if pull == PULL_DDEBS:
ddebs_mirror = config.get_value(distro.upper() + '_DDEBS_MIRROR')
if ddebs_mirror:
mirrors.append(ddebs_mirror)
if mirrors:
Logger.debug("using mirrors %s", ", ".join(mirrors))
package = options.package
release = getattr(options, 'release', None)
version = getattr(options, 'version', None)
pocket = None
dscfile = None
if package.endswith('.dsc') and not release and not version:
dscfile = package
package = None
if release:
try: try:
(release, pocket) = parse_release(release, distro) cls(*args, **kwargs).pull()
return
except KeyboardInterrupt:
Logger.normal('User abort.')
except (PackageNotFoundException, SeriesNotFoundException,
PocketDoesNotExistError, InvalidDistroValueError) as e:
Logger.error(str(e))
sys.exit(errno.ENOENT)
def __init__(self, *args, **kwargs):
self._default_pull = kwargs.get('pull')
self._default_distro = kwargs.get('distro')
self._default_arch = kwargs.get('arch', host_architecture())
self._parser = None
@property
def argparser(self):
if self._parser:
return self._parser
help_default_pull = "What to pull: " + ", ".join(VALID_PULLS)
if self._default_pull:
help_default_pull += (" (default: %s)" % self._default_pull)
help_default_distro = "Pull from: " + ", ".join(VALID_DISTROS)
if self._default_distro:
help_default_distro += (" (default: %s)" % self._default_distro)
help_default_arch = ("Get binary packages for arch")
help_default_arch += ("(default: %s)" % self._default_arch)
parser = ArgumentParser()
parser.add_argument('-v', '--verbose', action='store_true',
help="Print verbose/debug messages")
parser.add_argument('-d', '--download-only', action='store_true',
help="Do not extract the source package")
parser.add_argument('-m', '--mirror', action='append',
help='Preferred mirror(s)')
parser.add_argument('--no-conf', action='store_true',
help="Don't read config files or environment variables")
parser.add_argument('--no-verify-signature', action='store_true',
help="Don't fail if dsc signature can't be verified")
parser.add_argument('-a', '--arch', default=self._default_arch,
help=help_default_arch)
parser.add_argument('-p', '--pull', default=self._default_pull,
help=help_default_pull)
parser.add_argument('-D', '--distro', default=self._default_distro,
help=help_default_distro)
parser.add_argument('package', help="Package name to pull")
parser.add_argument('release', nargs='?', help="Release to pull from")
parser.add_argument('version', nargs='?', help="Package version to pull")
self._parser = parser
return self._parser
def parse_pull(self, pull):
if not pull:
raise InvalidPullValueError("Must specify --pull")
# allow 'dbgsym' as alias for 'ddebs'
if pull == 'dbgsym':
Logger.debug("Pulling '%s' for '%s'", PULL_DDEBS, pull)
pull = PULL_DDEBS
# assume anything starting with 'bin' means 'debs'
if str(pull).startswith('bin'):
Logger.debug("Pulling '%s' for '%s'", PULL_DEBS, pull)
pull = PULL_DEBS
# verify pull action is valid
if pull not in VALID_PULLS:
raise InvalidPullValueError("Invalid pull action '%s'" % pull)
return pull
def parse_distro(self, distro):
if not distro:
raise InvalidDistroValueError("Must specify --distro")
distro = distro.lower()
# allow 'lp' for 'ubuntu'
if distro == 'lp':
Logger.debug("Using distro '%s' for '%s'", DISTRO_UBUNTU, distro)
distro = DISTRO_UBUNTU
# assume anything with 'cloud' is UCA
if re.match(r'.*cloud.*', distro):
Logger.debug("Using distro '%s' for '%s'", DISTRO_UCA, distro)
distro = DISTRO_UCA
# verify distro is valid
if distro not in VALID_DISTROS:
raise InvalidDistroValueError("Invalid distro '%s'" % distro)
return distro
def parse_release(self, distro, release):
if distro == DISTRO_UCA:
# UCA is special; it is specified UBUNTURELEASE-UCARELEASE or just
# UCARELEASE. The user could also specify UCARELEASE-POCKET. But UCA
# archives always correspond to only one UBUNTURELEASE, and UCA archives
# have only the Release pocket, so only UCARELEASE matters to us.
for r in release.split('-'):
if r in UbuntuCloudArchiveSourcePackage.getReleaseNames():
Logger.debug("Using UCA release '%s'", r)
return (r, None)
raise SeriesNotFoundException('UCA release {} not found.'.format(release))
# Check if release[-pocket] is specified
(release, pocket) = split_release_pocket(release, default=None)
Logger.debug("Parsed release '%s' pocket '%s'", release, pocket)
if distro == DISTRO_DEBIAN:
# This converts from the aliases like 'unstable'
debian_info = DebianDistroInfo()
codename = debian_info.codename(release)
if codename:
Logger.normal("Using release '%s' for '%s'", codename, release)
release = codename
d = Distribution(distro)
# let SeriesNotFoundException flow up
d.getSeries(release)
Logger.debug("Using distro '%s' release '%s' pocket '%s'",
distro, release, pocket)
return (release, pocket)
def parse_release_and_version(self, distro, release, version, try_swap=True):
# Verify specified release is valid, and params in correct order
pocket = None
try:
(release, pocket) = self.parse_release(distro, release)
except (SeriesNotFoundException, PocketDoesNotExistError): except (SeriesNotFoundException, PocketDoesNotExistError):
Logger.debug("Param '%s' not valid series, must be version", release) if try_swap:
release, version = version, release Logger.debug("Param '%s' not valid series, must be version", release)
if release: release, version = version, release
try: if release:
(release, pocket) = parse_release(release, distro) return self.parse_release_and_version(distro, release, version, False)
except (SeriesNotFoundException, PocketDoesNotExistError): else:
Logger.error("Can't find series for '%s' or '%s'", Logger.error("Can't find series for '%s' or '%s'", release, version)
release, version) raise
raise return (release, version, pocket)
try: def parse_options(self, options):
pkgcls = DISTRO_PKG_CLASS[distro] # if any of these fail, there is a problem with the parser
srcpkg = pkgcls(package=package, version=version, # they should all be provided, though the optional ones may be None
series=release, pocket=pocket,
mirrors=mirrors, dscfile=dscfile, # type bool
verify_signature=(not options.no_verify_signature)) assert 'download_only' in options
assert 'no_conf' in options
assert 'no_verify_signature' in options
# type string
assert 'pull' in options
assert 'distro' in options
assert 'arch' in options
assert 'package' in options
# type string, optional
assert 'release' in options
assert 'version' in options
# type list of strings, optional
assert 'mirror' in options
pull = self.parse_pull(options['pull'])
distro = self.parse_distro(options['distro'])
params = {}
params['package'] = options['package']
if options['release']:
(r, v, p) = self.parse_release_and_version(distro, options['release'],
options['version'])
params['series'] = r
params['version'] = v
params['pocket'] = p
if (params['package'].endswith('.dsc') and not params['series'] and not params['version']):
params['dscfile'] = params['package']
params.pop('package')
mirrors = []
if options['mirror']:
mirrors.append(options['mirror'])
if pull == PULL_DDEBS:
config = UDTConfig(options['no_conf'])
ddebs_mirror = config.get_value(distro.upper() + '_DDEBS_MIRROR')
if ddebs_mirror:
mirrors.append(ddebs_mirror)
if mirrors:
Logger.debug("using mirrors %s", ", ".join(mirrors))
params['mirrors'] = mirrors
params['verify_signature'] = not options['no_verify_signature']
return (pull, distro, params)
def pull(self, args=None):
"""Pull (download) specified package file(s)"""
options = vars(self.argparser.parse_args(args))
assert 'verbose' in options
if options['verbose'] is not None:
Logger.set_verbosity(options['verbose'])
Logger.debug("pullpkg options: %s", options)
# Login anonymously to LP
Launchpad.login_anonymously()
(pull, distro, params) = self.parse_options(options)
# call implementation, and allow exceptions to flow up to caller
srcpkg = DISTRO_PKG_CLASS[distro](**params)
spph = srcpkg.lp_spph spph = srcpkg.lp_spph
except PackageNotFoundException as e:
Logger.error(str(e))
raise
Logger.normal('Found %s', spph.display_name) Logger.normal('Found %s', spph.display_name)
if pull == PULL_LIST: if pull == PULL_LIST:
Logger.normal("Source files:") Logger.normal("Source files:")
for f in srcpkg.dsc['Files']: for f in srcpkg.dsc['Files']:
Logger.normal(" %s", f['name']) Logger.normal(" %s", f['name'])
Logger.normal("Binary files:") Logger.normal("Binary files:")
for f in spph.getBinaries(options.arch): for f in spph.getBinaries(options['arch']):
Logger.normal(" %s", f.getFileName()) Logger.normal(" %s", f.getFileName())
return elif pull == PULL_SOURCE:
# allow DownloadError to flow up to caller
# allow DownloadError to flow up to caller srcpkg.pull()
if pull == PULL_SOURCE: if options['download_only']:
srcpkg.pull() Logger.debug("--download-only specified, not extracting")
if options.download_only: else:
Logger.debug("--download-only specified, not extracting") srcpkg.unpack()
else: else:
srcpkg.unpack() name = '.*'
else: if params['package'] != spph.getPackageName():
name = '.*' Logger.normal("Pulling only binary package '%s'", params['package'])
if package != spph.getPackageName(): Logger.normal("Use package name '%s' to pull all binary packages",
Logger.normal("Pulling only binary package '%s'", package) spph.getPackageName())
Logger.normal("Use package name '%s' to pull all binary packages", name = params['package']
spph.getPackageName()) if pull == PULL_DEBS:
name = package name = r'{}(?<!-di)(?<!-dbgsym)$'.format(name)
if pull == PULL_DEBS: elif pull == PULL_DDEBS:
name = r'{}(?<!-di)(?<!-dbgsym)$'.format(name) name += '-dbgsym$'
elif pull == PULL_DDEBS: elif pull == PULL_UDEBS:
name += '-dbgsym$' name += '-di$'
elif pull == PULL_UDEBS: else:
name += '-di$' raise InvalidPullValueError("Invalid pull value %s" % pull)
else:
raise InvalidPullValueError("Invalid pull value %s" % pull) # allow DownloadError to flow up to caller
total = srcpkg.pull_binaries(name=name, arch=options.arch) total = srcpkg.pull_binaries(name=name, arch=options['arch'])
if total < 1: if total < 1:
Logger.error("No %s found for %s %s", pull, Logger.error("No %s found for %s %s", pull,
package, spph.getVersion()) params['package'], spph.getVersion())