#-----------------------------------------------------------------------------# Copyright (c) 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 importsfromtypingimportTYPE_CHECKING,Any# Bokeh importsfrom...core.serializationimportSerializerfrom...document.callbacksimportinvoke_with_curdocfrom...document.jsonimportPatchJsonfrom..messageimportMessageifTYPE_CHECKING:from...core.has_propsimportSetterfrom...document.documentimportDocumentfrom...document.eventsimportDocumentPatchedEvent#-----------------------------------------------------------------------------# Globals and constants#-----------------------------------------------------------------------------__all__=('patch_doc',)#-----------------------------------------------------------------------------# 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],**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")[doc]=docsserializer=Serializer(references=doc.models.synced_references)patch_json=PatchJson(events=serializer.encode(events))doc.models.flush_synced(lambdamodel:notserializer.has_ref(model))msg=cls(header,metadata,patch_json)forbufferinserializer.buffers:msg.add_buffer(buffer)returnmsg