f.seek(start)
if comp_type.startswith(b'gzip'):
gzip = try_import('gzip')
- return cast(bytes, gzip.open(f).read(size))
+ data = f.read(size)
+ return cast(bytes, gzip.decompress(data))
elif comp_type.startswith(b'lz4'):
lz4 = try_import('lz4.frame', 'lz4')
- return cast(bytes, lz4.frame.decompress(f.read(size)))
+ data = f.read(size)
+ return cast(bytes, lz4.frame.decompress(data))
elif comp_type.startswith(b'lzma'):
lzma = try_import('lzma')
- return cast(bytes, lzma.open(f).read(size))
+ data = f.read(size)
+ return cast(bytes, lzma.decompress(data))
elif comp_type.startswith(b'lzo'):
raise NotImplementedError('lzo decompression not implemented')
elif comp_type.startswith(b'xzkern'):
raise NotImplementedError('xzkern decompression not implemented')
elif comp_type.startswith(b'zstd'):
zstd = try_import('zstandard')
- return cast(bytes, zstd.ZstdDecompressor().stream_reader(f.read(size)).read())
+ data = f.read(size)
+ return cast(bytes, zstd.ZstdDecompressor().stream_reader(data).read())
raise NotImplementedError(f'unknown compressed type: {comp_type!r}')