def _fetch()

in src/buildstream_plugins/sources/git.py [0:0]

• 48 lines of code
• 10 McCabe index (conditional complexity)

    def _fetch(self, url, fetch_all=False):
        self._ensure_repo()

        # Work out whether we can fetch a specific tag: are we given a ref which
        # 1. is in git-describe format
        # 2. refers to an exact tag (is "...-0-g...")
        # 3. is available on the remote and tags the specified commit?
        # And lastly: are we on a new-enough Git which allows cloning from our potentially shallow cache?
        if fetch_all:
            pass
        # Fetching from a shallow-cloned repo was first supported in v1.9.0
        elif not self.ref or self.source.host_git_version is not None and self.source.host_git_version < (1, 9, 0):
            fetch_all = True
        else:
            m = re.match(EXACT_TAG_PATTERN, self.ref)
            if m is None:
                fetch_all = True
            else:
                tag = m.group("tag")
                commit = m.group("commit")

                if not self.remote_has_tag(url, tag, commit):
                    self.source.status(
                        "{}: {} is not advertised on {}. Fetching all Git refs".format(self.source, self.ref, url)
                    )
                    fetch_all = True
                else:
                    exit_code = self.source.call(
                        [
                            self.source.host_git,
                            "fetch",
                            "--depth=1",
                            url,
                            "+refs/tags/{tag}:refs/tags/{tag}".format(tag=tag),
                        ],
                        cwd=self.mirror,
                    )
                    if exit_code != 0:
                        self.source.status(
                            "{}: Failed to fetch tag '{}' from {}. Fetching all Git refs".format(self.source, tag, url)
                        )
                        fetch_all = True

        if fetch_all:
            self.source.call(
                [
                    self.source.host_git,
                    "fetch",
                    "--prune",
                    url,
                    "+refs/heads/*:refs/heads/*",
                    "+refs/tags/*:refs/tags/*",
                ],
                fail="Failed to fetch from remote git repository: {}".format(url),
                fail_temporarily=True,
                cwd=self.mirror,
            )