def stream_length()

in src/aws_encryption_sdk/streaming_client.py [0:0]

• 10 lines of code
• 5 McCabe index (conditional complexity)

    def stream_length(self):
        """Returns the length of the source stream, determining it if not already known."""
        if self._stream_length is None:
            try:
                current_position = self.source_stream.tell()
                self.source_stream.seek(0, 2)
                self._stream_length = self.source_stream.tell()
                self.source_stream.seek(current_position, 0)
            except Exception as error:
                # Catch-all for unknown issues encountered trying to seek for stream length
                raise NotSupportedError(error)
        return self._stream_length