Method write

python-client/src/datapane/_vendor/base64io/__init__.py:192–226

Base64-encode the bytes and write them to the wrapped stream. Any bytes that would require padding for the next write call are buffered until the next write or close. .. warning:: Because up to two bytes of data must be buffered to ensure correct base64 encodin

(self, b)

        return self.__wrapped.flush()

    def write(self, b):
        # type: (bytes) -> int
        """Base64-encode the bytes and write them to the wrapped stream.

        Any bytes that would require padding for the next write call are buffered until the
        next write or close.

        .. warning::

            Because up to two bytes of data must be buffered to ensure correct base64 encoding
            of all data written, this object **must** be closed after you are done writing to
            avoid data loss. If used as a context manager, we take care of that for you.

        :param bytes b: Bytes to write to wrapped stream
        :raises ValueError: if called on closed Base64IO object
        :raises IOError: if underlying stream is not writable
        """
        if self.closed:
            raise ValueError("I/O operation on closed file.")

        if not self.writable():
            raise IOError("Stream is not writable")

        # Load any stashed bytes and clear the buffer
        _bytes_to_write = self.__write_buffer + b
        self.__write_buffer = b""

        # If an even base64 chunk or finalizing the stream, write through.
        if len(_bytes_to_write) % 3 == 0:
            return self.__wrapped.write(base64.b64encode(_bytes_to_write))

        # We're not finalizing the stream, so stash the trailing bytes and encode the rest.
        trailing_byte_pos = -1 * (len(_bytes_to_write) % 3)
        self.__write_buffer = _bytes_to_write[trailing_byte_pos:]
        return self.__wrapped.write(base64.b64encode(_bytes_to_write[:trailing_byte_pos]))

    def writelines(self, lines):
        # type: (Iterable[bytes]) -> None