Source code for bumper

import argparse
import logging
import os
import sys

from bumper.cars import RequirementsBumper, RequirementsManager, BumpAccident
from bumper.utils import parse_requirements

log = logging.getLogger(__name__)


[docs]def bump(): """ CLI entry point to bump requirements in requirements.txt or pinned.txt """ parser = argparse.ArgumentParser(description=bump.__doc__) parser.add_argument('names', nargs='*', help=""" Only bump dependencies that match the name. Name can be a product group name defined in workspace.cfg. To bump to a specific version instead of latest, append version to name (i.e. requests==1.2.3 or 'requests>=1.2.3'). When > or < is used, be sure to quote.""") parser.add_argument('--add', '--require', action='store_true', help='Add the `names` to the requirements file if they don\'t exist.') parser.add_argument('--file', help='Requirement file to bump. Defaults to requirements.txt and pinned.txt') parser.add_argument('--force', action='store_true', help='Force a bump even when certain bump requirements are not met.') parser.add_argument('-d', '--detail', '--dependencies', action='store_true', help='If available, show detailed changes. For pinned.txt, pin parsed dependency requirements from changes') parser.add_argument('-n', '--dry-run', action='store_true', help='Perform a dry run without making changes') parser.add_argument('--debug', action='store_true', help='Turn on debug mode') args = parser.parse_args() targets = [args.file] if args.file else ['requirements.txt', 'pinned.txt'] level = logging.DEBUG if args.debug else logging.INFO logging.basicConfig(level=level, format='[%(levelname)s] %(message)s') try: bumper = BumperDriver(targets, full_throttle=args.force, detail=args.detail, test_drive=args.dry_run) bumper.bump(args.names, required=args.add, show_detail=args.detail) except Exception as e: if args.debug: raise else: log.error(e) sys.exit(1)
[docs]class BumperDriver(object): """ Driver that controls the main logic / coordinates the bumps with different bumper models (cars) """ def __init__(self, targets, bumper_models=None, full_throttle=False, detail=False, test_drive=False): """ :param list targets: List of file paths to bump :param list bumper_models: List of bumper classes that implements :class:`bumper.cars.AbstractBumper` :param bool full_throttle: Force bumps even when required requirements are not met :param bool detail: Generate detailed changes from changelog if possible. :param bool test_drive: Perform a dry run """ self.targets = targets self.bumper_models = bumper_models or [RequirementsBumper] self.full_throttle = full_throttle self.detail = detail self.test_drive = test_drive self.bumps = {} self.bumpers = []
[docs] def bump(self, filter_requirements, required=False, show_summary=True, show_detail=False, **kwargs): """ Bump dependency requirements using filter. :param list filter_requirements: List of dependency filter requirements. :param bool required: Require the filter_requirements to be met (by adding if possible). :param bool show_summary: Show summary for each bump made. :param bool show_detail: Show detail for each bump made if available. :return: Tuple with two elements: Dict of target file to bump message, List of :class:`Bump` :raise BumpAccident: for any bump errors """ found_targets = [target for target in self.targets if os.path.exists(target)] if not found_targets: raise BumpAccident('None of the requirement file(s) were found: %s' % ', '.join(self.targets)) bump_reqs = RequirementsManager() if filter_requirements: requirements = parse_requirements(filter_requirements) bump_reqs.add(requirements, required=required) try: for target in found_targets: log.debug('Target: %s', target) target_bumpers = [] target_bump_reqs = RequirementsManager(bump_reqs) loops = 0 while True: # Insurance to ensure that we don't get stuck forever. loops += 1 if loops > 5: log.debug('Too many transitive bump loops. Bailing out.') break if not target_bumpers: target_bumpers = [model(target, detail=self.detail, test_drive=self.test_drive) for model in self.bumper_models if model.likes(target)] if not target_bumpers: log.warn('No bumpers found that can bump %s', target) continue self.bumpers.extend(target_bumpers) new_target_bump_reqs = RequirementsManager() for bumper in target_bumpers: target_bumps = bumper.bump(target_bump_reqs) self.bumps.update(dict((b.name, b) for b in target_bumps)) for bump in target_bumps: for new_req in bump.requirements: if not (bump_reqs.satisfied_by_checked(new_req) or target_bump_reqs.satisfied_by_checked(new_req)): new_target_bump_reqs.add(new_req) bump_reqs.matched_name |= target_bump_reqs.matched_name bump_reqs.checked.extend(target_bump_reqs.checked) if new_target_bump_reqs: bump_reqs.add(new_target_bump_reqs) target_bump_reqs = RequirementsManager(list(r for r in new_target_bump_reqs if r.project_name not in self.bumps)) if not target_bump_reqs: break if not self.bumpers: raise BumpAccident('No bumpers found for %s' % ', '.join(found_targets)) if bump_reqs and not bump_reqs.matched_name: raise BumpAccident('None of the provided filter names were found in %s' % ', '.join(found_targets)) if self.bumps: for bump in self.bumps.values(): bump_reqs.check(bump) for reqs in bump_reqs.required_requirements().values(): for req in reqs: if not self.full_throttle: use_force = 'Use --force to ignore / force the bump' if req.required_by else '' raise BumpAccident('Requirement "%s" could not be met so bump can not proceed. %s' % (req, use_force)) if self.test_drive: log.info("Changes that would be made:\n") messages = {} for bumper in self.bumpers: if bumper.bumps: if not self.test_drive: bumper.update_requirements() if self.test_drive or show_summary: msg = bumper.bump_message(self.test_drive or show_detail) if self.test_drive: print msg else: rewords = [('Bump ', 'Bumped '), ('Pin ', 'Pinned '), ('Require ', 'Updated requirements: ')] for word, new_word in rewords: if msg.startswith(word): msg = msg.replace(word, new_word, 1) break log.info(msg) messages[bumper.target] = bumper.bump_message(True) return messages, self.bumps else: log.info('No need to bump. Everything is up to date!') return {}, [] except Exception: if not self.test_drive and self.bumps: map(lambda b: b.reverse(), self.bumpers) raise
[docs] def reverse(self): """ Reverse all bumpers """ if not self.test_drive and self.bumps: map(lambda b: b.reverse(), self.bumpers)