def fetch()

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

• 47 lines of code
• 15 McCabe index (conditional complexity)

    def fetch(self):
        # pylint: disable=arguments-differ

        with self.timed_activity(
            "Fetching image {}:{} with digest {}".format(self.image, self.tag, self.digest),
            silent_nested=True,
        ):
            with self.tempdir() as tmpdir:
                # move all files to a tmpdir
                try:
                    manifest = self._load_manifest()
                except FileNotFoundError as e:
                    try:
                        manifest_text, digest = self.client.manifest(
                            self.image,
                            self.digest,
                            self.architecture,
                            self.os,
                        )
                    except requests.RequestException as ee:
                        raise SourceError(ee) from ee

                    if digest != self.digest:
                        raise SourceError(
                            "Requested image {}, got manifest with digest {}".format(self.digest, digest)
                        ) from e
                    self._save_manifest(manifest_text, tmpdir)
                    manifest = json.loads(manifest_text)
                except DockerManifestError as e:
                    self.log("Unexpected manifest", detail=e.manifest)
                    raise
                except (OSError, requests.RequestException) as e:
                    raise SourceError(e) from e

                for layer in manifest["layers"]:
                    if layer["mediaType"] != "application/vnd.docker.image.rootfs.diff.tar.gzip":
                        raise SourceError("Unsupported layer type: {}".format(layer["mediaType"]))

                    layer_digest = layer["digest"]
                    blob_path = os.path.join(tmpdir, layer_digest + ".tar.gz")

                    if not os.path.exists(blob_path):
                        try:
                            self.client.blob(self.image, layer_digest, download_to=blob_path)
                        except (OSError, requests.RequestException) as e:
                            if os.path.exists(blob_path):
                                shutil.rmtree(blob_path)
                            raise SourceError(e) from e

                    self._verify_blob(blob_path, expected_digest=layer_digest)

                # Only if all sources are successfully fetched, move files to staging directory

                # As both the manifest and blobs are content addressable, we can optimize space by having
                # a flat mirror directory. We check one-by-one if there is any need to copy a file out of the tmpdir.
                for fetched_file in os.listdir(tmpdir):
                    move_atomic(
                        os.path.join(tmpdir, fetched_file),
                        os.path.join(self.get_mirror_directory(), fetched_file),
                    )