Skip to content

comm

Classes:

  • Comm –

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

Comm

Bases: BaseComm

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

Notes
  • 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.
Referenced by:

Methods:

Source code in src/async_kernel/comm.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class Comm(BaseComm):
    """
    An implementation of `comm.BaseComms` for async-kernel  ([on pypi](https://pypi.org/project/comm/)).

    Notes:
        - `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.
    """

    kernel: Kernel | None = None

    @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,
        )

    @override
    def handle_msg(self, msg: comm.base_comm.MessageType) -> None:
        """Handle a comm_msg message"""
        if self._msg_callback:
            self._msg_callback(msg)

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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@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 
55
56
57
58
59
@override
def handle_msg(self, msg: comm.base_comm.MessageType) -> None:
    """Handle a comm_msg message"""
    if self._msg_callback:
        self._msg_callback(msg)