def _detect_compression_type()

in torchdata/datapipes/iter/util/extractor.py [0:0]

• 21 lines of code
• 7 McCabe index (conditional complexity)

    def _detect_compression_type(self, path: str) -> CompressionType:
        if self.file_type:
            return self.file_type

        ext = "".join(pathlib.Path(path).suffixes)
        if ext in {".tar.gz", ".tar.xz"}:
            return self.types.TAR
        else:
            ext = os.path.splitext(path)[1]
        if ext == ".tar":
            return self.types.TAR
        elif ext == ".xz":
            return self.types.LZMA
        elif ext == ".gz":
            return self.types.GZIP
        elif ext == ".zip":
            return self.types.ZIP
        else:
            raise RuntimeError(
                f"File at {path} has file extension {ext}, which does not match what are supported by"
                f"ExtractorIterDataPipe."
            )