Method _write_compressed_data

apps/knowledge/models/knowledge.py:420–432

(self, data, block_size=64 * 1024)

        return result["lo_id"]

    def _write_compressed_data(self, data, block_size=64 * 1024):
        buffer = io.BytesIO(data)
        offset = 0

        while True:
            chunk = buffer.read(block_size)
            if not chunk:
                break

            offset += len(chunk)
            select_one(
                "SELECT lo_put(%s::oid, %s::bigint, %s::bytea)::VARCHAR;", [self.loid, offset - len(chunk), chunk]
            )

    def get_bytes(self):
        buffer = io.BytesIO()