def get_artifacts()

in scripts/metric_reporter/gcs_client.py [0:0]

• 18 lines of code
• 6 McCabe index (conditional complexity)

    def get_artifacts(self) -> list[GCSArtifacts]:
        """Get artifact files from GCS, grouped by repository.

        Returns:
            List[GCSArtifactPaths]: A list of grouped GCS paths per repository.

        Raises:
            GCSClientError: If the structure of the test_result_bucket is unexpected
        """
        # Here we assume that the structure of the test_result_bucket is as follows:
        # test_result_bucket/
        #     ├── repository/
        #          ├── coverage_artifact_dir/
        #               ├── coverage-1.json
        #               ├── coverage-2.json
        #          ├── junit_artifact_dir/
        #               ├── junit-1.xml
        #               ├── junit-2.xml
        artifact_map = {}  # Using a regular dictionary
        for blob in self._client.list_blobs(self._bucket):
            parts = blob.name.rstrip("/").split("/")
            if len(parts) != 3:
                continue
            repository, artifact_dir, artifact_file = parts

            if repository not in artifact_map:
                artifact_map[repository] = GCSArtifacts()

            if artifact_dir == self.coverage_artifact_dir:
                artifact_map[repository].coverage_artifact_files.append(artifact_file)
            elif artifact_dir == self.junit_artifact_dir:
                artifact_map[repository].junit_artifact_files.append(artifact_file)
            else:
                raise GCSClientError(
                    f"The artifact directory name {artifact_dir} from {blob.name} is unsupported."
                )

        return list(artifact_map.values())