Created
November 9, 2022 10:15
-
-
Save jelmervdl/c42cb53371a9c2a849076de6e58a9b5f to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import sys | |
| import contextlib | |
| import subprocess | |
| @contextlib.contextmanager | |
| def gzip(path, mode='r'): | |
| """Like gzip.open(), but using external gzip process which for some reason | |
| is a lot faster on macOS.""" | |
| try: | |
| fh = open(path, mode + 'b') # or check if path is not already a file-like obj? | |
| if mode == 'r': | |
| args = ['-cd', path] | |
| direction = 'stdin', 'stdout' | |
| elif mode == 'w': | |
| args = ['-c'] | |
| direction = 'stdout', 'stdin' | |
| else: | |
| raise ValueError('mode') | |
| kwargs = { | |
| direction[0]: fh, | |
| direction[1]: subprocess.PIPE | |
| } | |
| with subprocess.Popen(['gzip', *args], **kwargs) as proc: | |
| yield getattr(proc, direction[1]) | |
| getattr(proc, direction[1]).close() # close stdin when writing so gzip knows we're finished | |
| if proc.wait() != 0: | |
| raise RuntimeError(f'gzip returned error code {proc.returncode}') | |
| finally: | |
| if fh: | |
| fh.close() | |
| with gzip('file.gz', 'w') as fh: | |
| for n in range(100): | |
| fh.write((b'x' * n) + b"\n") | |
| with gzip('file.gz') as fh: | |
| # fh is BytesIO, might need to wrap it with TextIOWrapper to get unicode. | |
| for line in fh: | |
| sys.stdout.buffer.write(line) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment