improved documentation and error messages of CliCompressor

composer/utils/compression.py

@@ -26,7 +26,27 @@ def is_compressed_pt(filename: str) -> bool:
 
 
 class CliCompressor:
-    """Base class for data compression CLI tools."""
+    """Base class for data compression CLI tools.
+
+    This class handles compression and decompression of data by piping it through
+    CLI compressor tools installed on the system. e.g. the `gzip` command for producing `.gz` files.
+
+    Example:
+    .. code-block:: python
+
+        compressor = CliCompressor('gz', 'gzip')
+
+        with compressor.compress('myfile.txt.gz') as f:
+            f.write('foo')
+
+        with compressor.decompress('myfile.txt.gz') as f:
+            assert f.read() == 'foo'
+
+    Args:
+        extension (str): The suffix used to identify files that the compressor supports (without a leading `.`).
+        cmd (str, optional): The name of the CLI tool that this compressor uses. Defaults to `None`, in which case
+            it is assumed that the tool name is the same as the extension.
+    """
 
     def __init__(self, extension: str, cmd: Optional[str] = None) -> None:
         self.extension = extension
@@ -37,6 +57,7 @@ def __repr__(self) -> str:
 
     @property
     def exists(self) -> bool:
+        """Whether the CLI tool used by this compressor can be found."""
         return shutil.which(self.cmd) is not None
 
     def check_exists(self) -> None:
@@ -47,9 +68,10 @@ def _compress_cmd(self) -> List[str]:
         return [self.cmd]
 
     @contextmanager
-    def compress(self, filename: str) -> Iterator[IO[bytes]]:
+    def compress(self, out_filename: str) -> Iterator[IO[bytes]]:
+        """Compress some data, saving to the given file."""
         self.check_exists()
-        with open(filename, 'wb') as f:
+        with open(out_filename, 'wb') as f:
             proc = subprocess.Popen(
                 self._compress_cmd(),
                 stdin=subprocess.PIPE,
@@ -60,13 +82,14 @@ def compress(self, filename: str) -> Iterator[IO[bytes]]:
             proc.stdin.close()
             returncode = proc.wait()
             if returncode != 0:
-                raise IOError(f'failed to compress "{filename}" using {self!r}')
+                raise IOError(f'failed to compress to "{out_filename}" using {self!r} (return code {returncode})')
 
     def _decompress_cmd(self, filename: str) -> List[str]:
         return [self.cmd, '-dc', filename]
 
     @contextmanager
     def decompress(self, in_filename: str) -> Iterator[IO[bytes]]:
+        """Decompress the content of the given file, providing the output as a file-like object."""
         self.check_exists()
         proc = subprocess.Popen(
             self._decompress_cmd(in_filename),
@@ -76,7 +99,7 @@ def decompress(self, in_filename: str) -> Iterator[IO[bytes]]:
         yield proc.stdout
         returncode = proc.wait()
         if returncode != 0:
-            raise IOError(f'failed to decompress "{in_filename}" using {self!r}')
+            raise IOError(f'failed to decompress "{in_filename}" using {self!r} (return code {returncode})')
 
 
 def get_compressor(filename: str) -> CliCompressor: