def _flush_write_buffer()

in src/azstoragetorch/io.py [0:0]

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

    def _flush_write_buffer(self) -> None:
        if self._write_buffer:
            futures = self._client.stage_blocks(memoryview(self._write_buffer))
            self._all_stage_block_futures.extend(futures)
            self._in_progress_stage_block_futures.extend(futures)
            self._write_buffer = bytearray()