def compress_stream()

in azure-kusto-ingest/azure/kusto/ingest/descriptors.py [0:0]

• 14 lines of code
• 2 McCabe index (conditional complexity)

    def compress_stream(self) -> None:
        stream = self.stream
        zipped_stream = BytesIO()
        stream_buffer = stream.read()
        with GzipFile(filename="data", fileobj=zipped_stream, mode="wb") as f_out:
            if isinstance(stream_buffer, str):
                data = bytes(stream_buffer, "utf-8")
                f_out.write(data)
            else:
                f_out.write(stream_buffer)
        zipped_stream.seek(0)
        self.is_compressed = True
        self.stream_name += ".gz"
        self.stream = zipped_stream