Class ZstdFile

Lib/compression/zstd/_zstdfile.py:20–292

A file-like object providing transparent Zstandard (de)compression. A ZstdFile can act as a wrapper for an existing file object, or refer directly to a named file on disk. ZstdFile provides a *binary* file interface. Data is read and returned as bytes, and may only be written to ob

class ZstdFile(_streams.BaseStream):
    """A file-like object providing transparent Zstandard (de)compression.

    A ZstdFile can act as a wrapper for an existing file object, or refer
    directly to a named file on disk.

    ZstdFile provides a *binary* file interface. Data is read and returned as
    bytes, and may only be written to objects that support the Buffer Protocol.
    """

    FLUSH_BLOCK = ZstdCompressor.FLUSH_BLOCK
    FLUSH_FRAME = ZstdCompressor.FLUSH_FRAME

    def __init__(self, file, /, mode='r', *,
                 level=None, options=None, zstd_dict=None):
        """Open a Zstandard compressed file in binary mode.

        *file* can be either an file-like object, or a file name to open.

        *mode* can be 'r' for reading (default), 'w' for (over)writing, 'x' for
        creating exclusively, or 'a' for appending. These can equivalently be
        given as 'rb', 'wb', 'xb' and 'ab' respectively.

        *level* is an optional int specifying the compression level to use,
        or COMPRESSION_LEVEL_DEFAULT if not given.

        *options* is an optional dict for advanced compression parameters.
        See CompressionParameter and DecompressionParameter for the possible
        options.

        *zstd_dict* is an optional ZstdDict object, a pre-trained Zstandard
        dictionary. See train_dict() to train ZstdDict on sample data.
        """
        self._fp = None
        self._close_fp = False
        self._mode = _MODE_CLOSED
        self._buffer = None

        if not isinstance(mode, str):
            raise ValueError('mode must be a str')
        if options is not None and not isinstance(options, dict):
            raise TypeError('options must be a dict or None')
        mode = mode.removesuffix('b')  # handle rb, wb, xb, ab
        if mode == 'r':
            if level is not None:
                raise TypeError('level is illegal in read mode')
            self._mode = _MODE_READ
        elif mode in {'w', 'a', 'x'}:
            if level is not None and not isinstance(level, int):
                raise TypeError('level must be int or None')
            self._mode = _MODE_WRITE
            self._compressor = ZstdCompressor(level=level, options=options,
                                              zstd_dict=zstd_dict)
            self._pos = 0
        else:
            raise ValueError(f'Invalid mode: {mode!r}')

        if isinstance(file, (str, bytes, PathLike)):