def _consume_from_readline_buffer()

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

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

    def _consume_from_readline_buffer(self, consumed: bytes, limit: int) -> bytes:
        limit -= len(consumed)
        find_pos = self._readline_buffer.find(self._READLINE_TERMINATOR, 0, limit)
        end = find_pos + 1
        if find_pos == -1:
            buffer_length = len(self._readline_buffer)
            end = min(buffer_length, limit)
        consumed += self._readline_buffer[:end]
        self._readline_buffer = self._readline_buffer[end:]
        self._position += end
        return consumed