def consume_with_stream()

in confidence-shared/shared.py [0:0]

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

    def consume_with_stream(self):
        """Context-manager to consume a buffer using a _UniffiRustBufferStream.

        The _UniffiRustBuffer will be freed once the context-manager exits, ensuring that we don't
        leak it even if an error occurs.
        """
        try:
            s = _UniffiRustBufferStream.from_rust_buffer(self)
            yield s
            if s.remaining() != 0:
                raise RuntimeError("junk data left in buffer at end of consume_with_stream")
        finally:
            self.free()