Source code for bokeh.protocol.messages.patch_doc

from __future__ import absolute_import

from json import loads

from ...core.json_encoder import serialize_json
from ...document.util import references_json
from ..message import Message
from . import register

@register
[docs]class patch_doc_1(Message): ''' Define the ``PATCH-DOC`` message (revision 1) for sending Document patch events between remote documents. The ``content`` fragment of for this message is has the form: .. code-block:: python { 'events' : <protocol document events> 'references' : <model references> } ''' msgtype = 'PATCH-DOC' revision = 1 def __init__(self, header, metadata, content): super(patch_doc_1, self).__init__(header, metadata, content) @classmethod
[docs] def create(cls, events, use_buffers=True, **metadata): ''' Create a ``PATCH-DOC`` message Args: events (list) : A list of patch events to apply to a document Any additional keyword arguments will be put into the message ``metadata`` fragment as-is. ''' header = cls.create_header() if not events: raise ValueError("PATCH-DOC message requires at least one event") docs = { event.document for event in events } if len(docs) != 1: raise ValueError("PATCH-DOC message configured with events for more than one document") # this roundtrip is fortunate, but is needed because there are type conversions # in BokehJSONEncoder which keep us from easily generating non-string JSON patch_json, buffers = process_document_events(events, use_buffers) content = loads(patch_json) msg = cls(header, metadata, content) for (header, payload) in buffers: msg.add_buffer(header, payload) return msg
[docs] def apply_to_document(self, doc, setter=None): ''' ''' doc.apply_json_patch(self.content, setter)
[docs]def process_document_events(events, use_buffers=True): ''' Create a JSON string describing a patch to be applied as well as any optional buffers. Args: events : list of events to be translated into patches Returns: str, list : JSON string which can be applied to make the given updates to obj as well as any optional buffers ''' json_events = [] references = set() buffers = [] if use_buffers else None for event in events: json_events.append(event.generate(references, buffers)) json = { 'events' : json_events, 'references' : references_json(references), } return serialize_json(json), buffers if use_buffers else []