Method _read_with_offset

apps/knowledge/models/knowledge.py:453–468

()

    def get_bytes_stream(self, start=0, end=None, chunk_size=64 * 1024):
        def _read_with_offset():
            offset = start
            while True:
                result = select_one(
                    "SELECT lo_get(%s::oid, %s, %s) as chunk",
                    [self.loid, offset, end - offset if end and (end - offset) < chunk_size else chunk_size],
                )
                chunk = result["chunk"] if result else None
                if not chunk:
                    break
                yield chunk
                offset += len(chunk)
                if len(chunk) < chunk_size:
                    break
                if end and offset > end:
                    break

        return _read_with_offset()