def _by_lines_fallback()

in storage/decorator.py [0:0]

• 18 lines of code
• 3 McCabe index (conditional complexity)

    def _by_lines_fallback(json_collector_state: JsonCollectorState) -> StorageDecoratorIterator:
        @by_lines
        def wrapper(
            storage: ProtocolStorageType, range_start: int, body: BytesIO, is_gzipped: bool
        ) -> StorageDecoratorIterator:
            data_to_yield: bytes = body.read()
            yield data_to_yield, 0, range_start, b"", None

        for line, _, _, newline, _ in wrapper(
            json_collector_state.storage,
            json_collector_state.ending_offset,
            BytesIO(json_collector_state.unfinished_line),
            False,
        ):
            assert isinstance(line, bytes)

            _handle_offset(len(line) + len(newline), json_collector_state)

            # let's reset the buffer
            json_collector_state.unfinished_line = b""

            # let's set the flag for direct yield from now on
            json_collector_state.has_an_object_start = False

            yield line, _, _, newline, None