1  """Base class for a Comm"""
   2  
   3  # Copyright (c) IPython Development Team.
   4  # Distributed under the terms of the Modified BSD License.
   5  
   6  import uuid
   7  from typing import Optional
   8  from warnings import warn
   9  
  10  import comm.base_comm
  11  import traitlets.config
  12  from traitlets import Bool, Bytes, Instance, Unicode, default
  13  
  14  from ipykernel.jsonutil import json_clean
  15  from ipykernel.kernelbase import Kernel
  16  
  17  
  18  # this is the class that will be created if we do comm.create_comm
  19  class BaseComm(comm.base_comm.BaseComm):  # type:ignore[misc]
  20      """The base class for comms."""
  21  
  22      kernel: Optional["Kernel"] = None
  23  
  24      def publish_msg(self, msg_type, data=None, metadata=None, buffers=None, **keys):
  25          """Helper for sending a comm message on IOPub"""
  26          if not Kernel.initialized():
  27              return
  28  
  29          data = {} if data is None else data
  30          metadata = {} if metadata is None else metadata
  31          content = json_clean(dict(data=data, comm_id=self.comm_id, **keys))
  32  
  33          if self.kernel is None:
  34              self.kernel = Kernel.instance()
  35  
  36          assert self.kernel.session is not None
  37          self.kernel.session.send(
  38              self.kernel.iopub_socket,
  39              msg_type,
  40              content,
  41              metadata=json_clean(metadata),
  42              parent=self.kernel.get_parent(),
  43              ident=self.topic,
  44              buffers=buffers,
  45          )
  46  
  47  
  48  # but for backwards compatibility, we need to inherit from LoggingConfigurable
  49  class Comm(BaseComm, traitlets.config.LoggingConfigurable):
  50      """Class for communicating between a Frontend and a Kernel"""
  51  
  52      kernel = Instance("ipykernel.kernelbase.Kernel", allow_none=True)  # type:ignore[assignment]
  53      comm_id = Unicode()
  54      primary = Bool(True, help="Am I the primary or secondary Comm?")
  55  
  56      target_name = Unicode("comm")
  57      target_module = Unicode(
  58          None,
  59          allow_none=True,
  60          help="""requirejs module from
  61          which to load comm target.""",
  62      )
  63  
  64      topic = Bytes()
  65  
  66      @default("kernel")
  67      def _default_kernel(self):
  68          if Kernel.initialized():
  69              return Kernel.instance()
  70          return None
  71  
  72      @default("comm_id")
  73      def _default_comm_id(self):
  74          return uuid.uuid4().hex
  75  
  76      def __init__(
  77          self, target_name="", data=None, metadata=None, buffers=None, show_warning=True, **kwargs
  78      ):
  79          """Initialize a comm."""
  80          if show_warning:
  81              warn(
  82                  "The `ipykernel.comm.Comm` class has been deprecated. Please use the `comm` module instead."
  83                  "For creating comms, use the function `from comm import create_comm`.",
  84                  DeprecationWarning,
  85                  stacklevel=2,
  86              )
  87  
  88          # Handle differing arguments between base classes.
  89          had_kernel = "kernel" in kwargs
  90          kernel = kwargs.pop("kernel", None)
  91          if target_name:
  92              kwargs["target_name"] = target_name
  93          BaseComm.__init__(self, data=data, metadata=metadata, buffers=buffers, **kwargs)  # type:ignore[call-arg]
  94          # only re-add kernel if explicitly provided
  95          if had_kernel:
  96              kwargs["kernel"] = kernel
  97          traitlets.config.LoggingConfigurable.__init__(self, **kwargs)
  98  
  99  
 100  __all__ = ["Comm"]
 101