def _push_artifact_proto()

in src/buildstream/_artifactcache.py [0:0]

• 33 lines of code
• 13 McCabe index (conditional complexity)

    def _push_artifact_proto(self, element, artifact, artifact_digest, remote):

        artifact_proto = artifact._get_proto()

        keys = list(utils._deduplicate([artifact_proto.strong_key, artifact_proto.weak_key]))
        artifact_names = [element.get_artifact_name(key=key) for key in keys]
        uris = [REMOTE_ASSET_ARTIFACT_URN_TEMPLATE.format(artifact_name) for artifact_name in artifact_names]

        try:
            response = remote.fetch_blob(uris)
            # Skip push if artifact is already on the server
            if response and response.blob_digest == artifact_digest:
                return False
        except AssetCacheError as e:
            raise ArtifactError("{}".format(e), temporary=True) from e

        referenced_directories = []
        if artifact_proto.files:
            referenced_directories.append(artifact_proto.files)
        if artifact_proto.buildtree:
            referenced_directories.append(artifact_proto.buildtree)
        if artifact_proto.sources:
            referenced_directories.append(artifact_proto.sources)
        if artifact_proto.buildroot:
            referenced_directories.append(artifact_proto.buildroot)

        referenced_blobs = [artifact_proto.low_diversity_meta, artifact_proto.high_diversity_meta] + [
            log_file.digest for log_file in artifact_proto.logs
        ]

        try:
            remote.push_blob(
                uris,
                artifact_digest,
                references_blobs=referenced_blobs,
                references_directories=referenced_directories,
            )
        except AssetCacheError as e:
            raise ArtifactError("{}".format(e), temporary=True) from e

        return True