#-----------------------------------------------------------------------------# Copyright (c) 2012 - 2022, Anaconda, Inc., and Bokeh Contributors.# All rights reserved.## The full license is in the file LICENSE.txt, distributed with this software.#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Boilerplate#-----------------------------------------------------------------------------from__future__importannotationsimportlogging# isort:skiplog=logging.getLogger(__name__)#-----------------------------------------------------------------------------# Imports#-----------------------------------------------------------------------------# Standard library importsfromjsonimportloadsfromtypingimport(TYPE_CHECKING,Any,List,Set,Tuple,)# Bokeh importsfrom...core.json_encoderimportserialize_jsonfrom...document.callbacksimportinvoke_with_curdocfrom...document.jsonimportPatchJsonfrom...document.utilimportreferences_jsonfrom..messageimportMessageifTYPE_CHECKING:from...core.has_propsimportSetterfrom...document.documentimportDocumentfrom...document.eventsimportDocumentPatched,DocumentPatchedEventfrom...modelimportModelfrom..messageimportBufferRef#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=('patch_doc','process_document_events',)#-----------------------------------------------------------------------------# General API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Dev API#-----------------------------------------------------------------------------
[docs]classpatch_doc(Message[PatchJson]):''' Define the ``PATCH-DOC`` message 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'
[docs]@classmethoddefcreate(cls,events:List[DocumentPatchedEvent],use_buffers:bool=True,**metadata:Any)->patch_doc:''' 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()ifnotevents:raiseValueError("PATCH-DOC message requires at least one event")docs={event.documentforeventinevents}iflen(docs)!=1:raiseValueError("PATCH-DOC message configured with events for more than one document")# this roundtrip is unfortunate, but is needed because there are type conversions# in BokehJSONEncoder which keep us from easily generating non-string JSONpatch_json,buffers=process_document_events(events,use_buffers)content=loads(patch_json)msg=cls(header,metadata,content)for(buffer_header,payload)inbuffers:msg.add_buffer(buffer_header,payload)returnmsg
defprocess_document_events(events:List[DocumentPatchedEvent],use_buffers:bool=True)->Tuple[str,List[BufferRef]]:''' 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:List[DocumentPatched]=[]references:Set[Model]=set()buffers:List[BufferRef]|None=[]ifuse_bufferselseNoneforeventinevents:json_events.append(event.generate(references,buffers))json=PatchJson(events=json_events,references=references_json(references),)returnserialize_json(json),buffersor[]#-----------------------------------------------------------------------------# Private API#-----------------------------------------------------------------------------#-----------------------------------------------------------------------------# Code#-----------------------------------------------------------------------------