Skip to content

async_kernel.comm

Classes:

  • Comm –

    An implementation of comm.BaseComms for async-kernel (on pypi).

async_kernel.comm.Comm

Bases: BaseComm

An implementation of comm.BaseComms for async-kernel (on pypi).

Note

  • kernel is added/removed by the CommManager.
  • kernel is added to the CommManager by the kernel once the sockets have been opened.
  • publish_msg is no-op when kernel is unset.

Methods:

publish_msg 

publish_msg(
    msg_type: str,
    data: MaybeDict = None,
    metadata: MaybeDict = None,
    buffers: BuffersType = None,
    **keys,
)

Helper for sending a comm message on IOPub.

Source code in src/async_kernel/comm.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
@override
def publish_msg(
    self,
    msg_type: str,
    data: MaybeDict = None,
    metadata: MaybeDict = None,
    buffers: BuffersType = None,
    **keys,
):
    """Helper for sending a comm message on IOPub."""
    if (kernel := self.kernel) is None:
        # Only send when the kernel is set
        return
    content = {"data": {} if data is None else data, "comm_id": self.comm_id} | keys
    kernel.iopub_send(
        msg_or_type=msg_type,
        content=content,
        metadata=metadata,
        parent=None,
        ident=self.topic,
        buffers=buffers,
    )

handle_msg 

handle_msg(msg: MessageType) -> None

Handle a comm_msg message

Source code in src/async_kernel/comm.py 
64
65
66
67
68
@override
def handle_msg(self, msg: comm.base_comm.MessageType) -> None:
    """Handle a comm_msg message"""
    if self._msg_callback:
        self._msg_callback(msg)

async_kernel.comm.CommManager 

CommManager()

Bases: HasTraits, CommManager

The comm manager for all Comm instances.

Note

  • When the trait CommManager.kernel is set the Comm.kernel trait is set on all async_kernel.comm.Comm instances.
  • The Comm will only send messages when the kernel is set.
  • The kernel sets CommManager.kernel when its ready the iopub socket is open.

Methods:

Source code in src/async_kernel/comm.py 
93
94
def __init__(self) -> None:
    super().__init__()

register_comm 

register_comm(comm: BaseComm) -> str

Register a new comm

Source code in src/async_kernel/comm.py 
103
104
105
106
107
108
@override
def register_comm(self, comm: comm.base_comm.BaseComm) -> str:
    """Register a new comm"""
    if isinstance(comm, Comm) and (kernel := self.kernel):
        comm.kernel = kernel
    return super().register_comm(comm)

async_kernel.comm.set_comm 

set_comm()

Monkey patch the comm module's functions to provide iopub comms.

  1. comm.create_comm to async_kernel.comm.Comm.
  2. comm.get_com_manager to async_kernel.comm.CommManager.
Source code in src/async_kernel/comm.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def set_comm():
    """
    Monkey patch the [comm](https://pypi.org/project/comm/) module's functions to provide iopub comms.

    1.  `comm.create_comm` to [async_kernel.comm.Comm][].
    1. `comm.get_com_manager` to [async_kernel.comm.CommManager][].
    """
    comm.create_comm = Comm
    comm.get_comm_manager = get_comm_manager

    # # Monkey patch ipykernel for modules that use it such as pyviz_comms:https://github.com/holoviz/pyviz_comms/blob/4cd44d902364590ba8892c8e7f48d7888d0a1c0c/pyviz_comms/__init__.py#L403C14-L403C28
    with contextlib.suppress(ImportError):
        import ipykernel.comm  # noqa: PLC0415

        ipykernel.comm.Comm = Comm
        ipykernel.comm.CommManager = CommManager