Compress some data, saving to the given file.
(self, out_filename: str)
| 69 | |
| 70 | @contextmanager |
| 71 | def compress(self, out_filename: str) -> Iterator[IO[bytes]]: |
| 72 | """Compress some data, saving to the given file.""" |
| 73 | self.check_exists() |
| 74 | with open(out_filename, 'wb') as f: |
| 75 | proc = subprocess.Popen( |
| 76 | self._compress_cmd(), |
| 77 | stdin=subprocess.PIPE, |
| 78 | stdout=f, |
| 79 | ) |
| 80 | assert proc.stdin is not None |
| 81 | yield proc.stdin |
| 82 | proc.stdin.close() |
| 83 | returncode = proc.wait() |
| 84 | if returncode != 0: |
| 85 | raise IOError(f'failed to compress to "{out_filename}" using {self!r} (return code {returncode})') |
| 86 | |
| 87 | def _decompress_cmd(self, filename: str) -> list[str]: |
| 88 | return [self.cmd, '-dc', filename] |