def __init__()

in src/python/pants/backend/jvm/tasks/jar_publish.py [0:0]

• 83 lines of code
• 21 McCabe index (conditional complexity)

  def __init__(self, *args, **kwargs):
    super(JarPublish, self).__init__(*args, **kwargs)
    self.cachedir = os.path.join(self.workdir, 'cache')

    self._jvm_options = self.get_options().jvm_options

    self.log = self.context.log

    if self.get_options().local:
      local_repo = dict(
        resolver='publish_local',
        path=os.path.abspath(os.path.expanduser(self.get_options().local)),
        confs=['default'],
        auth=None
      )
      self.repos = defaultdict(lambda: local_repo)
      self.commit = False
      self.local_snapshot = self.get_options().local_snapshot
    else:
      self.repos = self.get_options().repos
      if not self.repos:
        raise TaskError(
          "This repo is not configured to publish externally! Please configure per\n"
          "http://pantsbuild.org/publish.html#authenticating-to-the-artifact-repository,\n"
          "by setting --publish-jar-repos=<dict> or re-run with '--publish-jar-local=<dir>'.")
      for repo, data in self.repos.items():
        auth = data.get('auth')
        if auth:
          credentials = next(iter(self.context.resolve(auth)))
          user = credentials.username(data['resolver'])
          password = credentials.password(data['resolver'])
          self.context.log.debug('Found auth for repo={} user={}'.format(repo, user))
          self.repos[repo]['username'] = user
          self.repos[repo]['password'] = password
      self.commit = self.get_options().commit
      self.push_postscript = self.get_options().push_postscript or ''
      self.local_snapshot = False

    self.scm = get_scm() if self.commit else None

    self.named_snapshot = self.get_options().named_snapshot
    if self.named_snapshot:
      self.named_snapshot = Namedver.parse(self.named_snapshot)

    self.dryrun = self.get_options().dryrun
    self.force = self.get_options().force
    self.publish_changelog = self.get_options().changelog and self.scm

    def parse_jarcoordinate(coordinate):
      components = coordinate.split('#', 1)
      if len(components) == 2:
        org, name = components
        return org, name
      else:
        spec = components[0]
        address = Address.parse(spec)
        try:
          self.context.build_graph.inject_address_closure(address)
          target = self.context.build_graph.get_target(address)
          if not target:
            siblings = self.context.address_mapper.addresses_in_spec_path(address.spec_path)
            prompt = 'did you mean' if len(siblings) == 1 else 'maybe you meant one of these'
            raise TaskError('{} => {}?:\n    {}'.format(address, prompt,
                                                        '\n    '.join(str(a) for a in siblings)))
          if not self._is_exported(target):
            raise TaskError('{} is not an exported target'.format(coordinate))
          return target.provides.org, target.provides.name
        except (BuildFile.BuildFileError,
                BuildFileParser.BuildFileParserError,
                AddressLookupError) as e:
          raise TaskError('{message}\n  Problem identifying target at {spec}'
                          .format(message=e, spec=spec))

    self.overrides = {}
    if self.get_options().override:
      if self.named_snapshot:
        raise TaskError('Options --named-snapshot and --override are mutually exclusive!')

      def parse_override(override):
        try:
          coordinate, rev = override.split('=', 1)
          try:
            # overrides imply semantic versioning
            rev = Semver.parse(rev)
          except ValueError as e:
            raise TaskError('Invalid version {}: {}'.format(rev, e))
          return parse_jarcoordinate(coordinate), rev
        except ValueError:
          raise TaskError('Invalid override: {}'.format(override))

      self.overrides.update(parse_override(o) for o in self.get_options().override)

    self.restart_at = None
    if self.get_options().restart_at:
      self.restart_at = parse_jarcoordinate(self.get_options().restart_at)