Method try_pull

misc/python/materialize/mzbuild.py:1135–1187

Download the image if it does not exist locally. Returns whether it was found.

(self, max_retries: int)

                time.sleep(sleep_time)

    def try_pull(self, max_retries: int) -> bool:
        """Download the image if it does not exist locally. Returns whether it was found."""
        command = ["docker", "pull"]
        # --quiet skips printing the progress bar, which does not display well in CI.
        if ui.env_is_truthy("CI"):
            command.append("--quiet")
        command.append(self.spec())
        if not self.acquired:
            ui.header(f"Acquiring {self.spec()}")
            sleep_time = 1
            for retry in range(1, max_retries + 1):
                try:
                    spawn.runv(
                        command,
                        stdin=subprocess.DEVNULL,
                        stdout=sys.stderr.buffer,
                    )
                    self.acquired = True
                    break
                except subprocess.CalledProcessError:
                    if retry < max_retries:
                        # There seems to be no good way to tell what error
                        # happened based on error code
                        # (https://github.com/docker/cli/issues/538) and we
                        # want to print output directly to terminal.
                        if build := os.getenv("CI_WAITING_FOR_BUILD"):
                            for retry in range(max_retries):
                                try:
                                    build_status = buildkite.get_build_status(build)
                                except subprocess.CalledProcessError:
                                    time.sleep(sleep_time)
                                    sleep_time = min(sleep_time * 2, 10)
                                    break
                                print(f"Build {build} status: {build_status}")
                                if build_status == "failed":
                                    print(
                                        f"Build {build} has been marked as failed, exiting hard"
                                    )
                                    sys.exit(1)
                                elif build_status == "success":
                                    break
                                assert (
                                    build_status == "pending"
                                ), f"Unknown build status {build_status}"
                                time.sleep(1)
                        else:
                            print(f"Retrying in {sleep_time}s ...")
                            time.sleep(sleep_time)
                            sleep_time = min(sleep_time * 2, 10)
                        continue
                    else:
                        break
        return self.acquired

    def is_published_if_necessary(self) -> bool:
        """Report whether the image exists on DockerHub & GHCR if it is publishable."""