Skip to content

messaging

Modules:

  • base –

    Default Messaging object definitions for Messaging including BaseMessage, Connection and Client.

  • zmq –

    ZMQ messaging objects using zmq sockets.

Classes:

  • LocalClient –

    A client for an interface running in the current process.

LocalClient

Bases: HasInterface[T_interface_co], BaseClient[T_interface_co], Generic[T_interface_co]

A client for an interface running in the current process.

Referenced by:

Attributes:

Source code in src/async_kernel/messaging/base.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
class LocalClient(HasInterface[T_interface_co], BaseClient[T_interface_co], Generic[T_interface_co]):
    """A client for an interface running in the current process."""

    connection: Fixed[Self, Connection[T_interface_co]] = Fixed(lambda c: Connection(session_id=c["owner"].session_id))
    """A local connection to the interface."""

    @override
    async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:
        await self.parent.started
        # Cross-connect
        self.connection.transmit_msg = self.handle_incoming_msg
        self.transmit_msg = self.connection.handle_incoming_msg

        async with self.connection.start():
            await super().connection_task(started, stop)

connection class-attribute instance-attribute 

connection: Fixed[Self, Connection[T_interface_co]] = Fixed(
    lambda c: Connection(session_id=c["owner"].session_id)
)

A local connection to the interface.

Default Messaging object definitions for Messaging including BaseMessage, Connection and Client.

Classes:

  • BaseMessage –

    The base for messaging between kernel interfaces and clients.

  • Connection –

    Provides a connection to the interface for messaging.

  • BaseClient –

    Communicates with a single connection.

  • LocalClient –

    A client for an interface running in the current process.

BaseMessage

Bases: StartStopTask, LoggingConfigurable, MessageProtocol

The base for messaging between kernel interfaces and clients.

Subclassed by:

Methods:

  • __init__ –

    Initialize the instance.

  • send_message –

    Sends the message to the other side (client for kernel and vice versa) and returns a PendingMessage.

Attributes:

Source code in src/async_kernel/messaging/base.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class BaseMessage(StartStopTask, LoggingConfigurable, MessageProtocol):
    """The base for messaging between kernel interfaces and clients."""

    session_id: Fixed[Self, str] = Fixed(lambda c: c["owner"]._session_id)
    """Used to identify this object as the `session` in a message header."""

    _pending_messages: Fixed[Self, dict[str, PendingMessage[Any]]] = Fixed(dict)
    """A mapping of the `msg_id` of message requests to the pending that is resolved with a reply."""

    def __init__(self, caller: Caller | None = None, /, session_id: str | NoValue = NoValue, **kwargs: Any) -> None:
        """Initialize the instance.

        Args:
            caller: The caller to use to run the interface.
            session_id: The id to use for to identify the instance in `msg["header"]["session"]`.
            **kwargs: Additional arguments to configure the instance.
        """
        super().__init__(**kwargs)
        # Set session using a temporary variable.
        self._session_id = str(uuid4()) if session_id is NoValue else session_id
        self.session_id  # noqa: B018
        del self._session_id
        self.set_task_function(self.connection_task, caller=caller or Caller())
        self.caller.protected = True

    async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:
        started()
        await stop

    @override
    def handle_reply(self, msg: Message) -> None:
        # Thread: undefined
        if (parent := msg.get("parent_header")) and (f := self._pending_messages.pop(parent["msg_id"], None)):
            self.log.debug("Received %s %r", msg["header"]["msg_type"], msg)
            f.set_result(msg)

    @property
    def as_owner(self) -> Callable[[], Self]:
        """Provides a callable with reference to self."""
        return lambda: self

    @override
    def msg(
        self,
        msg_type: str | MsgType,
        content: T | None,
        channel: Channel,
        *,
        parent: Message | dict[str, Any] | None = None,
        header: MsgHeader | dict[str, Any] | None = None,
        metadata: dict[str, Any] | None = None,
        buffers: BuffersType | None = None,
    ) -> Message[T]:
        parent = parent or utils.get_parent_message()
        if header is None:
            header = MsgHeader(
                date=datetime.now(tz=UTC),
                msg_id=str(uuid4()),
                msg_type=msg_type,
                session=self.session_id,
                username="",
                version=async_kernel.kernel_protocol_version,
            )
        return Message(
            channel=channel,
            header=header,  # pyright: ignore[reportArgumentType]
            parent_header=extract_header(parent),  # pyright: ignore[reportArgumentType]
            content={} if content is None else content,
            metadata=metadata if metadata is not None else {},
            buffers=[] if buffers is None else buffers,
        )

    @final
    def _base_send_msg(self, msg: Message, ident: bytes | list[bytes] | None = None) -> Message:
        self.transmit_msg(msg, [] if ident is None else ident if isinstance(ident, list) else [ident])
        return msg

    @override
    @final
    def send_message(
        self,
        msg: Message,
        ident: bytes | list[bytes] | None = None,
    ) -> PendingMessage[Content]:
        """Sends the message to the other side (client for kernel and vice versa) and returns a PendingMessage."""
        if MsgType(msg["header"]["msg_type"]) in MsgTypeNoReply:
            msg_ = f"{msg['header']['msg_type']} does not send a reply! Use `send_message_no_reply` instead."
            raise TypeError(msg_)
        self.log.debug("Send mssage %s %s", msg["header"]["msg_type"], msg)
        self._pending_messages[msg["header"]["msg_id"]] = pen = PendingMessage()
        pen.metadata.update(parent=self._base_send_msg(msg, ident))
        return pen

    @override
    @final
    def send_message_no_reply(self, msg: Message, ident: bytes | list[bytes] | None = None) -> None:
        self._base_send_msg(msg, ident)

    @override
    def send_reply(self, job: Job, content: dict, /, *, buffers: BuffersType | None = None) -> None:
        if "status" not in content:
            content["status"] = "ok"
        msg = self.msg(
            job["msg"]["header"]["msg_type"].replace("request", "reply"),
            content,
            job["msg"]["channel"],
            parent=job["msg"],
        )
        self.send_message_no_reply(msg, job["ident"])
        if msg:
            self.log.debug("send_reply %s", msg)

    @override
    def transmit_msg(self, msg: Message, ident: list[bytes]) -> None:
        raise NotImplementedError

session_id class-attribute instance-attribute 

session_id: Fixed[Self, str] = Fixed(lambda c: c['owner']._session_id)

Used to identify this object as the session in a message header.

as_owner property 

as_owner: Callable[[], Self]

Provides a callable with reference to self.

__init__ 

__init__(
    caller: Caller | None = None, /, session_id: str | NoValue = NoValue, **kwargs: Any
) -> None

Parameters:

  • caller

    (Caller | None, default: None ) –

    The caller to use to run the interface.

  • session_id

    (str | NoValue, default: NoValue ) –

    The id to use for to identify the instance in msg["header"]["session"].

  • **kwargs

    (Any, default: {} ) –

    Additional arguments to configure the instance.

Source code in src/async_kernel/messaging/base.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __init__(self, caller: Caller | None = None, /, session_id: str | NoValue = NoValue, **kwargs: Any) -> None:
    """Initialize the instance.

    Args:
        caller: The caller to use to run the interface.
        session_id: The id to use for to identify the instance in `msg["header"]["session"]`.
        **kwargs: Additional arguments to configure the instance.
    """
    super().__init__(**kwargs)
    # Set session using a temporary variable.
    self._session_id = str(uuid4()) if session_id is NoValue else session_id
    self.session_id  # noqa: B018
    del self._session_id
    self.set_task_function(self.connection_task, caller=caller or Caller())
    self.caller.protected = True

send_message 

send_message(
    msg: Message, ident: bytes | list[bytes] | None = None
) -> PendingMessage[Content]

Sends the message to the other side (client for kernel and vice versa) and returns a PendingMessage.

Source code in src/async_kernel/messaging/base.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@override
@final
def send_message(
    self,
    msg: Message,
    ident: bytes | list[bytes] | None = None,
) -> PendingMessage[Content]:
    """Sends the message to the other side (client for kernel and vice versa) and returns a PendingMessage."""
    if MsgType(msg["header"]["msg_type"]) in MsgTypeNoReply:
        msg_ = f"{msg['header']['msg_type']} does not send a reply! Use `send_message_no_reply` instead."
        raise TypeError(msg_)
    self.log.debug("Send mssage %s %s", msg["header"]["msg_type"], msg)
    self._pending_messages[msg["header"]["msg_id"]] = pen = PendingMessage()
    pen.metadata.update(parent=self._base_send_msg(msg, ident))
    return pen

Connection

Bases: HasInterface[T_interface_co], BaseMessage, Generic[T_interface_co]

Provides a connection to the interface for messaging.

Returned by: Subclassed by: Used by:

Methods:

  • connection_task –

    Open the channels, set ready when ready block until stopped, Don't call directly.

  • handle_incoming_msg –

    The handler for messages received on this connection.

  • iopub_send –

    Publish an iopub message.

Source code in src/async_kernel/messaging/base.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
class Connection(HasInterface[T_interface_co], BaseMessage, Generic[T_interface_co]):
    """Provides a connection to the interface for messaging."""

    @override
    async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:
        """Open the channels, set ready when ready block until stopped, Don't call directly."""
        self.parent.update_connections(self)
        started()
        await stop
        self.parent.update_connections()

    @override
    def handle_incoming_msg(self, msg: Message, ident: list[bytes]) -> None:
        """The handler for  messages received on this connection.

        Args:
            msg: A new message.
            ident: A list of bytes to route a reply message back to the origin. This can
                be an empty list when there is only one connection, such as [LocalClient][].
        """
        if msg["header"]["msg_type"].endswith("_reply"):
            self.handle_reply(msg)
        else:
            self.parent.kernel.handle_request(
                Job(msg=msg, ident=ident, received_time=time.monotonic(), owner=self.as_owner)
            )

    def connection_info(self) -> str:
        return ""

    def iopub_send(
        self,
        msg_type: IOPubMsgTypeAlias | str,
        content: Content | None = None,
        *,
        metadata: dict[str, Any] | None = None,
        parent: dict[str, Any] | MsgHeader | NoValue | None = NoValue,
        ident: bytes | list[bytes] | None = None,
        buffers: BuffersType | None = None,
    ) -> None:
        """Publish an iopub message."""
        self._base_send_msg(
            self.msg(
                MsgType(msg_type),
                content,
                Channel.iopub,
                parent=parent if parent is not NoValue else async_kernel.utils.get_parent_message(),  # pyright: ignore[reportArgumentType]
                metadata=metadata,
                buffers=buffers,
            ),
            ident,
        )
        self.log.debug("iopub_send: msg_type:%r %s", msg_type, msg_type)

connection_task async 

connection_task(started: Callable[[], Any], stop: ProtectedPending) -> None

Open the channels, set ready when ready block until stopped, Don't call directly.

Source code in src/async_kernel/messaging/base.py 
191
192
193
194
195
196
197
@override
async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:
    """Open the channels, set ready when ready block until stopped, Don't call directly."""
    self.parent.update_connections(self)
    started()
    await stop
    self.parent.update_connections()

handle_incoming_msg 

handle_incoming_msg(msg: Message, ident: list[bytes]) -> None

The handler for messages received on this connection.

Parameters:

  • msg

    (Message) –

    A new message.

  • ident

    (list[bytes]) –

    A list of bytes to route a reply message back to the origin. This can be an empty list when there is only one connection, such as LocalClient.

Source code in src/async_kernel/messaging/base.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
@override
def handle_incoming_msg(self, msg: Message, ident: list[bytes]) -> None:
    """The handler for  messages received on this connection.

    Args:
        msg: A new message.
        ident: A list of bytes to route a reply message back to the origin. This can
            be an empty list when there is only one connection, such as [LocalClient][].
    """
    if msg["header"]["msg_type"].endswith("_reply"):
        self.handle_reply(msg)
    else:
        self.parent.kernel.handle_request(
            Job(msg=msg, ident=ident, received_time=time.monotonic(), owner=self.as_owner)
        )

iopub_send 

iopub_send(
    msg_type: IOPubMsgTypeAlias | str,
    content: Content | None = None,
    *,
    metadata: dict[str, Any] | None = None,
    parent: dict[str, Any] | MsgHeader | NoValue | None = NoValue,
    ident: bytes | list[bytes] | None = None,
    buffers: BuffersType | None = None,
) -> None

Publish an iopub message.

Source code in src/async_kernel/messaging/base.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def iopub_send(
    self,
    msg_type: IOPubMsgTypeAlias | str,
    content: Content | None = None,
    *,
    metadata: dict[str, Any] | None = None,
    parent: dict[str, Any] | MsgHeader | NoValue | None = NoValue,
    ident: bytes | list[bytes] | None = None,
    buffers: BuffersType | None = None,
) -> None:
    """Publish an iopub message."""
    self._base_send_msg(
        self.msg(
            MsgType(msg_type),
            content,
            Channel.iopub,
            parent=parent if parent is not NoValue else async_kernel.utils.get_parent_message(),  # pyright: ignore[reportArgumentType]
            metadata=metadata,
            buffers=buffers,
        ),
        ident,
    )
    self.log.debug("iopub_send: msg_type:%r %s", msg_type, msg_type)

BaseClient

Bases: BaseMessage, Generic[T_interface_co]

Communicates with a single connection.

Subclassed by:

Methods:

  • input_request –

    Handle an input_request raised by the connected kernel.

  • iopub_subscribe –

    Open a new iopub socket and subscribe to a particular topic.

  • execute –

    Execute code in the kernel.

  • complete –

    Tab complete text in the kernel's namespace.

  • inspect –

    Get metadata information about an object in the kernel's namespace.

  • history –

    Get entries from the kernel's history list.

  • kernel_info –

    Request kernel info.

  • comm_info –

    Request comm info.

  • is_complete –

    Ask the kernel whether some code is complete and ready to execute.

  • shutdown –

    Request an immediate kernel shutdown.

Attributes:

Source code in src/async_kernel/messaging/base.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
class BaseClient(BaseMessage, Generic[T_interface_co]):
    """Communicates with a single connection."""

    _input_handlers: Fixed[Self, dict[str, Callable[[Content], CoroutineType[Any, Any, str]]]] = Fixed(dict)

    default_input_hander: Callable[[Content], CoroutineType[Any, Any, str]] | None = traitlets.Callable(  # pyright: ignore[reportAssignmentType]
        None, allow_none=True
    ).tag(config=True)
    """The default handler for input requests."""

    _iopub_queues: Fixed[Self, deque[tuple[bytes, SingleAsyncQueue]]] = Fixed(deque)

    @override
    def handle_incoming_msg(self, msg: Message, ident: list[bytes]) -> None:
        if msg["channel"] is Channel.iopub:
            for topic, queue in self._iopub_queues:
                if not topic or any(topic == v[: len(topic)] for v in ident):
                    queue.append(msg)
        elif msg["header"]["msg_type"].endswith("_reply"):
            self.handle_reply(msg)
        else:
            self._handle_request(Job(owner=self.as_owner, msg=msg, ident=ident, received_time=time.monotonic()))

    def _handle_request(self, job: Job) -> None:
        # Thread: undefined
        self.log.debug("Client handler request  %s %r", job["msg"]["header"]["msg_type"], job["msg"])
        handler = getattr(self, job["msg"]["header"]["msg_type"])
        self.caller.to_thread(self._wrap_request_handler, handler, job)

    async def _wrap_request_handler(self, func: Callable[[Job], CoroutineType[Any, Any, Content]], job: Job) -> None:
        """Handle messages from the kernel (interface), currently only `input_request` is implemented."""
        reply_msg_type: MsgType = MsgType(job["msg"]["header"]["msg_type"].replace("request", "reply"))
        try:
            content = await func(job)
            assert content["status"] in ["error", "ok"]
        except Exception as e:
            content = utils.error_to_content(e)
        msg = self.msg(reply_msg_type, content, job["msg"]["channel"], parent=job["msg"])
        self.send_message_no_reply(msg, job["ident"])

    async def input_request(self, job: Job[Content]) -> Content:
        """Handle an `input_request` raised by the connected kernel."""
        if (parent := job["msg"]["parent_header"]) and (handler := self._input_handlers.pop(parent["msg_id"], None)):
            result = await handler(job["msg"]["content"])
            return Content(status="ok", value=result)
        msg_ = "A handler is not available!"
        raise RuntimeError(msg_)

    @asynccontextmanager
    async def iopub_subscribe(
        self, topic: bytes = b"", *, timeout: float | None = 1
    ) -> AsyncGenerator[SingleAsyncQueue[Message]]:
        """Open a new iopub socket and subscribe to a particular topic.

        Args:
            topic: The topics to subscribe to.
            timeout: The maximum time to wait for a welcome message.

        Raise:
            TimeoutError: If a welcome message is not received in time.

        Usaage:
        ```python
        async with client.iopub_subscribe() as queue:
            async for msg in queue:
                pass
        ```
        """
        queue = SingleAsyncQueue()
        self._iopub_queues.append((topic, queue))
        try:
            yield queue
        finally:
            self._iopub_queues.remove((topic, queue))
            queue.stop()

    # Methods to send specific messages on channels (only relevant to execute). All other message types are decided by the kernel.
    def execute(
        self,
        code: str,
        silent: bool = False,
        *,
        store_history: bool = True,
        user_expressions: dict[str, str] | None = None,
        stop_on_error: NoValue | bool = NoValue,
        metadata: dict[str, Any] | None = None,
        input_handler: Callable[[Content], CoroutineType[Any, Any, str]] | NoValue | None = NoValue,
        channel: Literal[Channel.shell, Channel.control] = Channel.shell,
        subshell_id: str | None = None,
    ) -> PendingMessage:
        """Execute code in the kernel.

        Params:
            code: A string of code in the kernel's language.
            silent: If set, the kernel will execute the code as quietly possible, and
                will force store_history to be False.
            store_history: If set, the kernel will store command history.  This is forced
                to be False if silent is True.
            user_expressions: A dict mapping names to expressions to be evaluated in the user's
                dict. The expression values are returned as strings formatted using [repr][].
            input_handler:  A handler for the stdin requests associated with the execute request.
                When not provided, stdin is disabled.
            stop_on_error: Flag whether to abort the execution queue, if an exception is encountered.
        """
        input_handler = self.default_input_hander if input_handler is NoValue else input_handler
        content: ExecuteContent = {
            "code": code,
            "silent": silent,
            "store_history": store_history,
            "user_expressions": user_expressions or {},
            "allow_stdin": bool(input_handler),
            "stop_on_error": (not silent) if stop_on_error is NoValue else stop_on_error,
            "subshell_id": subshell_id,
        }
        msg = self.msg(MsgType.execute_request, content, channel, metadata=metadata)
        if input_handler:
            self._input_handlers[msg["header"]["msg_id"]] = input_handler
        pen = self.send_message(msg)
        pen.add_done_callback(lambda _: self._input_handlers.pop(pen.msg_id, None))
        return pen

    def complete(self, code: str, cursor_pos: int | None = None) -> PendingMessage[Content]:
        """Tab complete text in the kernel's namespace.

        Args:
            code: The context in which completion is requested.
                Can be anything between a variable name and an entire cell.
            cursor_pos: The position of the cursor in the block of code where the completion was requested.
                Default: `len(code)`.
        """
        if cursor_pos is None:
            cursor_pos = len(code)
        content = {"code": code, "cursor_pos": cursor_pos}
        msg = self.msg(MsgType.complete_request, content, Channel.shell)
        return self.send_message(msg)

    def inspect(self, code: str, cursor_pos: int | None = None, detail_level: int = 0) -> PendingMessage[Content]:
        """Get metadata information about an object in the kernel's namespace.

        It is up to the kernel to determine the appropriate object to inspect.

        Params:
            code: Context in which info is requested.
                Can be anything between a variable name and an entire cell.
            cursor_pos: The position of the cursor in the block of code where the info was requested.
            detail_level:  The level of detail for the introspection (0-2).
        """
        if cursor_pos is None:
            cursor_pos = len(code)
        content = {"code": code, "cursor_pos": cursor_pos, "detail_level": detail_level}
        return self.send_message(self.msg(MsgType.inspect_request, content, Channel.shell))

    def history(
        self,
        raw: bool = True,
        output: bool = False,
        hist_access_type: Literal["tail", "range", "search"] = "range",
        **kwargs: Any,
    ) -> PendingMessage[Content]:
        """Get entries from the kernel's history list.

        Args:
        raw: If True, return the raw input.
        output: If True, then return the output as well.
        hist_access_type: 'range' (fill in session, start and stop params), 'tail' (fill in n)
             or 'search' (fill in pattern param).
        **kwargs:
            session: For a range request, the session from which to get lines. Session numbers
                are positive integers; negative ones count back from the current session.
            start: The first line number of a history range.
            stop: The final (excluded) line number of a history range.
            n: The number of lines of history to get for a tail request.
            pattern: The glob-syntax pattern for a search request.

        Returns: The ID of the message sent.
        """
        if hist_access_type == "range":
            kwargs.setdefault("session", 0)
            kwargs.setdefault("start", 0)
        content = dict(raw=raw, output=output, hist_access_type=hist_access_type, **kwargs)
        return self.send_message(self.msg(MsgType.history_request, content, Channel.shell))

    def kernel_info(self) -> PendingMessage[Content]:
        """Request kernel info."""
        return self.send_message(self.msg(MsgType.kernel_info_request, None, Channel.shell))

    def comm_info(self, target_name: str | None = None) -> PendingMessage[Content]:
        """Request comm info."""
        content = {} if target_name is None else {"target_name": target_name}
        return self.send_message(self.msg(MsgType.comm_info_request, content, Channel.shell))

    def is_complete(self, code: str) -> PendingMessage[Content]:
        """Ask the kernel whether some code is complete and ready to execute."""
        return self.send_message(self.msg(MsgType.is_complete_request, {"code": code}, Channel.shell))

    def shutdown(self, restart: bool = False) -> PendingMessage[Content]:
        """Request an immediate kernel shutdown.

        Upon receipt of the (empty) reply, client code can safely assume that
        the kernel has shut down and it's safe to forcefully terminate it if
        it's still alive.
        """
        return self.send_message(self.msg(MsgType.shutdown_request, {"restart": restart}, Channel.control))

default_input_hander class-attribute instance-attribute 

default_input_hander: Callable[[Content], CoroutineType[Any, Any, str]] | None = (
    traitlets.Callable(None, allow_none=True).tag(config=True)
)

The default handler for input requests.

input_request async 

input_request(job: Job[Content]) -> Content

Handle an input_request raised by the connected kernel.

Source code in src/async_kernel/messaging/base.py 
283
284
285
286
287
288
289
async def input_request(self, job: Job[Content]) -> Content:
    """Handle an `input_request` raised by the connected kernel."""
    if (parent := job["msg"]["parent_header"]) and (handler := self._input_handlers.pop(parent["msg_id"], None)):
        result = await handler(job["msg"]["content"])
        return Content(status="ok", value=result)
    msg_ = "A handler is not available!"
    raise RuntimeError(msg_)

iopub_subscribe async 

iopub_subscribe(
    topic: bytes = b"", *, timeout: float | None = 1
) -> AsyncGenerator[SingleAsyncQueue[Message]]

Open a new iopub socket and subscribe to a particular topic.

Parameters:

  • topic

    (bytes, default: b'' ) –

    The topics to subscribe to.

  • timeout

    (float | None, default: 1 ) –

    The maximum time to wait for a welcome message.

Raise

TimeoutError: If a welcome message is not received in time.

Usaage: 

async with client.iopub_subscribe() as queue:
    async for msg in queue:
        pass

Source code in src/async_kernel/messaging/base.py 
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@asynccontextmanager
async def iopub_subscribe(
    self, topic: bytes = b"", *, timeout: float | None = 1
) -> AsyncGenerator[SingleAsyncQueue[Message]]:
    """Open a new iopub socket and subscribe to a particular topic.

    Args:
        topic: The topics to subscribe to.
        timeout: The maximum time to wait for a welcome message.

    Raise:
        TimeoutError: If a welcome message is not received in time.

    Usaage:
    ```python
    async with client.iopub_subscribe() as queue:
        async for msg in queue:
            pass
    ```
    """
    queue = SingleAsyncQueue()
    self._iopub_queues.append((topic, queue))
    try:
        yield queue
    finally:
        self._iopub_queues.remove((topic, queue))
        queue.stop()

execute 

execute(
    code: str,
    silent: bool = False,
    *,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    stop_on_error: NoValue | bool = NoValue,
    metadata: dict[str, Any] | None = None,
    input_handler: Callable[[Content], CoroutineType[Any, Any, str]]
    | NoValue
    | None = NoValue,
    channel: Literal[shell, control] = shell,
    subshell_id: str | None = None,
) -> PendingMessage

Execute code in the kernel.

Parameters:

  • code

    (str) –

    A string of code in the kernel's language.

  • silent

    (bool, default: False ) –

    If set, the kernel will execute the code as quietly possible, and will force store_history to be False.

  • store_history

    (bool, default: True ) –

    If set, the kernel will store command history. This is forced to be False if silent is True.

  • user_expressions

    (dict[str, str] | None, default: None ) –

    A dict mapping names to expressions to be evaluated in the user's dict. The expression values are returned as strings formatted using repr.

  • input_handler

    (Callable[[Content], CoroutineType[Any, Any, str]] | NoValue | None, default: NoValue ) –

    A handler for the stdin requests associated with the execute request. When not provided, stdin is disabled.

  • stop_on_error

    (NoValue | bool, default: NoValue ) –

    Flag whether to abort the execution queue, if an exception is encountered.

Source code in src/async_kernel/messaging/base.py 
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def execute(
    self,
    code: str,
    silent: bool = False,
    *,
    store_history: bool = True,
    user_expressions: dict[str, str] | None = None,
    stop_on_error: NoValue | bool = NoValue,
    metadata: dict[str, Any] | None = None,
    input_handler: Callable[[Content], CoroutineType[Any, Any, str]] | NoValue | None = NoValue,
    channel: Literal[Channel.shell, Channel.control] = Channel.shell,
    subshell_id: str | None = None,
) -> PendingMessage:
    """Execute code in the kernel.

    Params:
        code: A string of code in the kernel's language.
        silent: If set, the kernel will execute the code as quietly possible, and
            will force store_history to be False.
        store_history: If set, the kernel will store command history.  This is forced
            to be False if silent is True.
        user_expressions: A dict mapping names to expressions to be evaluated in the user's
            dict. The expression values are returned as strings formatted using [repr][].
        input_handler:  A handler for the stdin requests associated with the execute request.
            When not provided, stdin is disabled.
        stop_on_error: Flag whether to abort the execution queue, if an exception is encountered.
    """
    input_handler = self.default_input_hander if input_handler is NoValue else input_handler
    content: ExecuteContent = {
        "code": code,
        "silent": silent,
        "store_history": store_history,
        "user_expressions": user_expressions or {},
        "allow_stdin": bool(input_handler),
        "stop_on_error": (not silent) if stop_on_error is NoValue else stop_on_error,
        "subshell_id": subshell_id,
    }
    msg = self.msg(MsgType.execute_request, content, channel, metadata=metadata)
    if input_handler:
        self._input_handlers[msg["header"]["msg_id"]] = input_handler
    pen = self.send_message(msg)
    pen.add_done_callback(lambda _: self._input_handlers.pop(pen.msg_id, None))
    return pen

complete 

complete(code: str, cursor_pos: int | None = None) -> PendingMessage[Content]

Tab complete text in the kernel's namespace.

Parameters:

  • code

    (str) –

    The context in which completion is requested. Can be anything between a variable name and an entire cell.

  • cursor_pos

    (int | None, default: None ) –

    The position of the cursor in the block of code where the completion was requested. Default: len(code).

Source code in src/async_kernel/messaging/base.py 
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def complete(self, code: str, cursor_pos: int | None = None) -> PendingMessage[Content]:
    """Tab complete text in the kernel's namespace.

    Args:
        code: The context in which completion is requested.
            Can be anything between a variable name and an entire cell.
        cursor_pos: The position of the cursor in the block of code where the completion was requested.
            Default: `len(code)`.
    """
    if cursor_pos is None:
        cursor_pos = len(code)
    content = {"code": code, "cursor_pos": cursor_pos}
    msg = self.msg(MsgType.complete_request, content, Channel.shell)
    return self.send_message(msg)

inspect 

inspect(
    code: str, cursor_pos: int | None = None, detail_level: int = 0
) -> PendingMessage[Content]

Get metadata information about an object in the kernel's namespace.

It is up to the kernel to determine the appropriate object to inspect.

Parameters:

  • code

    (str) –

    Context in which info is requested. Can be anything between a variable name and an entire cell.

  • cursor_pos

    (int | None, default: None ) –

    The position of the cursor in the block of code where the info was requested.

  • detail_level

    (int, default: 0 ) –

    The level of detail for the introspection (0-2).

Source code in src/async_kernel/messaging/base.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
def inspect(self, code: str, cursor_pos: int | None = None, detail_level: int = 0) -> PendingMessage[Content]:
    """Get metadata information about an object in the kernel's namespace.

    It is up to the kernel to determine the appropriate object to inspect.

    Params:
        code: Context in which info is requested.
            Can be anything between a variable name and an entire cell.
        cursor_pos: The position of the cursor in the block of code where the info was requested.
        detail_level:  The level of detail for the introspection (0-2).
    """
    if cursor_pos is None:
        cursor_pos = len(code)
    content = {"code": code, "cursor_pos": cursor_pos, "detail_level": detail_level}
    return self.send_message(self.msg(MsgType.inspect_request, content, Channel.shell))

history 

history(
    raw: bool = True,
    output: bool = False,
    hist_access_type: Literal["tail", "range", "search"] = "range",
    **kwargs: Any,
) -> PendingMessage[Content]

Get entries from the kernel's history list.

Args: raw: If True, return the raw input. output: If True, then return the output as well. hist_access_type: 'range' (fill in session, start and stop params), 'tail' (fill in n) or 'search' (fill in pattern param). **kwargs: session: For a range request, the session from which to get lines. Session numbers are positive integers; negative ones count back from the current session. start: The first line number of a history range. stop: The final (excluded) line number of a history range. n: The number of lines of history to get for a tail request. pattern: The glob-syntax pattern for a search request.

Returns: The ID of the message sent.

Source code in src/async_kernel/messaging/base.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def history(
    self,
    raw: bool = True,
    output: bool = False,
    hist_access_type: Literal["tail", "range", "search"] = "range",
    **kwargs: Any,
) -> PendingMessage[Content]:
    """Get entries from the kernel's history list.

    Args:
    raw: If True, return the raw input.
    output: If True, then return the output as well.
    hist_access_type: 'range' (fill in session, start and stop params), 'tail' (fill in n)
         or 'search' (fill in pattern param).
    **kwargs:
        session: For a range request, the session from which to get lines. Session numbers
            are positive integers; negative ones count back from the current session.
        start: The first line number of a history range.
        stop: The final (excluded) line number of a history range.
        n: The number of lines of history to get for a tail request.
        pattern: The glob-syntax pattern for a search request.

    Returns: The ID of the message sent.
    """
    if hist_access_type == "range":
        kwargs.setdefault("session", 0)
        kwargs.setdefault("start", 0)
    content = dict(raw=raw, output=output, hist_access_type=hist_access_type, **kwargs)
    return self.send_message(self.msg(MsgType.history_request, content, Channel.shell))

kernel_info 

kernel_info() -> PendingMessage[Content]

Request kernel info.

Source code in src/async_kernel/messaging/base.py 
425
426
427
def kernel_info(self) -> PendingMessage[Content]:
    """Request kernel info."""
    return self.send_message(self.msg(MsgType.kernel_info_request, None, Channel.shell))

comm_info 

comm_info(target_name: str | None = None) -> PendingMessage[Content]

Request comm info.

Source code in src/async_kernel/messaging/base.py 
429
430
431
432
def comm_info(self, target_name: str | None = None) -> PendingMessage[Content]:
    """Request comm info."""
    content = {} if target_name is None else {"target_name": target_name}
    return self.send_message(self.msg(MsgType.comm_info_request, content, Channel.shell))

is_complete 

is_complete(code: str) -> PendingMessage[Content]

Ask the kernel whether some code is complete and ready to execute.

Source code in src/async_kernel/messaging/base.py 
434
435
436
def is_complete(self, code: str) -> PendingMessage[Content]:
    """Ask the kernel whether some code is complete and ready to execute."""
    return self.send_message(self.msg(MsgType.is_complete_request, {"code": code}, Channel.shell))

shutdown 

shutdown(restart: bool = False) -> PendingMessage[Content]

Request an immediate kernel shutdown.

Upon receipt of the (empty) reply, client code can safely assume that the kernel has shut down and it's safe to forcefully terminate it if it's still alive.

Source code in src/async_kernel/messaging/base.py 
438
439
440
441
442
443
444
445
def shutdown(self, restart: bool = False) -> PendingMessage[Content]:
    """Request an immediate kernel shutdown.

    Upon receipt of the (empty) reply, client code can safely assume that
    the kernel has shut down and it's safe to forcefully terminate it if
    it's still alive.
    """
    return self.send_message(self.msg(MsgType.shutdown_request, {"restart": restart}, Channel.control))

LocalClient

Bases: HasInterface[T_interface_co], BaseClient[T_interface_co], Generic[T_interface_co]

A client for an interface running in the current process.

Referenced by:

Attributes:

Source code in src/async_kernel/messaging/base.py 
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
class LocalClient(HasInterface[T_interface_co], BaseClient[T_interface_co], Generic[T_interface_co]):
    """A client for an interface running in the current process."""

    connection: Fixed[Self, Connection[T_interface_co]] = Fixed(lambda c: Connection(session_id=c["owner"].session_id))
    """A local connection to the interface."""

    @override
    async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:
        await self.parent.started
        # Cross-connect
        self.connection.transmit_msg = self.handle_incoming_msg
        self.transmit_msg = self.connection.handle_incoming_msg

        async with self.connection.start():
            await super().connection_task(started, stop)

connection class-attribute instance-attribute 

connection: Fixed[Self, Connection[T_interface_co]] = Fixed(
    lambda c: Connection(session_id=c["owner"].session_id)
)

A local connection to the interface.

ZMQ messaging objects using zmq sockets.

Classes:

ZMQConnection

Bases: ZMQMessage, Connection[T_interface_co], Generic[T_interface_co]

Provides the ZMQ sockets for clients to connect and communicate with the interface.

Referenced by:
Source code in src/async_kernel/messaging/zmq.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class ZMQConnection(ZMQMessage, Connection[T_interface_co], Generic[T_interface_co]):
    """Provides the ZMQ sockets for clients to connect and communicate with the interface."""

    @property
    @override
    def kernel_name(self) -> str:  # pyright: ignore[reportIncompatibleVariableOverride]
        return self.parent.kernel_name

    @override
    async def connection_task(self, started: Callable[[], Any], stop: ProtectedPending) -> None:

        def heartbeat_handler(hb: ZMQPollSocket, event: int) -> None:
            # Thread: zmq_poll_thread
            hb.send_multipart(hb.recv_multipart())

        def iopub_reg_handler(socket: ZMQPollSocket, flags: int) -> None:
            """https://jupyter-client.readthedocs.io/en/stable/messaging.html#welcome-message."""
            # Thread: zmq_poll_thread
            # handle PUB subscribe/unsubscribe messages.
            # welcome_message:  https://jupyter.org/enhancement-proposals/65-jupyter-xpub/jupyter-xpub.html#replace-pub-socket-with-xpub-socket
            msg = socket.recv()
            if msg[0] == 1:
                ident = msg[1:]
                msg = self.msg(MsgType.iopub_welcome, {"subscription": ident.decode()}, Channel.iopub)
                self.session.send(socket, msg, ident=[ident])  # pyright: ignore[reportArgumentType]

        def handler(sock, event, channel: Channel, recv=self.session.recv, handle_msg=self.handle_incoming_msg) -> None:
            # Thread: zmq_poll_thread
            ident, msg = recv(sock)
            msg["channel"] = channel
            handle_msg(msg, ident)

        with self.zmq_poll as zpoll:
            await self._bind_sockets()
            self.parent.update_connections(self)
            started()
            await self.parent.started
            with (
                zpoll.event_handler(self._sockets[Channel.control], partial(handler, channel=Channel.control)),
                zpoll.event_handler(self._sockets[Channel.shell], functools.partial(handler, channel=Channel.shell)),
                zpoll.event_handler(self._sockets[Channel.stdin], functools.partial(handler, channel=Channel.stdin)),
                zpoll.event_handler(self._sockets[Channel.heartbeat], heartbeat_handler),
                zpoll.event_handler(self._sockets[Channel.iopub], iopub_reg_handler),
            ):
                await stop
                self.parent.update_connections()

    async def _bind_sockets(self):
        """Create, configure and bind all sockets."""

        def bind_sockets() -> None:
            if os.path.exists(self.connection_file):  # noqa: PTH110
                self.load_connection_file()
            self.write_connection_file()

            for channel in Channel:
                port = int(getattr(self, f"{channel}_port"))
                assert port
                if channel is not Channel.stdin:
                    assert channel not in self._sockets

                match channel:
                    case Channel.shell | Channel.control | Channel.heartbeat | Channel.stdin:
                        socket = self.zmq_poll.socket(zmq.SocketType.ROUTER)
                        # ref: https://github.com/ipython/ipykernel/issues/270
                        socket.router_handover = 1
                    case Channel.iopub:
                        socket = self.zmq_poll.socket(zmq.SocketType.XPUB)
                socket.setsockopt(zmq.SocketOption.LINGER, 500)
                socket.identity = self.session.bsession
                if self.curve_secretkey is not None:
                    socket.curve_secretkey = self.curve_secretkey
                    socket.curve_publickey = self.curve_publickey
                    socket.curve_server = True
                # Bind the socket.
                addr = f"tcp://{self.ip}:{port}" if self.transport == "tcp" else f"ipc://{self.ip}-{port}"
                self.log.debug("%s socket on port: %i", channel, port)
                self._sockets[channel] = socket
                socket.bind(addr)

        await self.zmq_poll.aexecute(bind_sockets)

    @override
    def connection_info(self) -> str:
        if self.connection_file and (f := pathlib.Path(self.connection_file)).exists():
            return f"connection_file: {f}\nInfo: {json.dumps(json.loads(f.read_bytes()), indent=2)}"
        return ""

ZMQClient

Bases: BaseClient[T_interface_co], ZMQMessage, Generic[T_interface_co]

A client for an interface that provides a ZMQConnection.

The client can be connected to an existing interface's using either: - ZMQClient.load_connection_info or, - ZMQClient.load_connection_file

A new interface/kernel can be started with ZMQClient.subprocess_kernel.

Methods:

  • start –

    Connect this client to the interface.

  • subprocess_kernel –

    Start a kernel interface as a subprocess.

Attributes:

Source code in src/async_kernel/messaging/zmq.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
class ZMQClient(BaseClient[T_interface_co], ZMQMessage, Generic[T_interface_co]):
    """A client for an interface that provides a [ZMQConnection][].

    The client can be connected to an existing interface's using either:
    - `ZMQClient.load_connection_info` or,
    - `ZMQClient.load_connection_file`

    A new interface/kernel can be started with [ZMQClient.subprocess_kernel][].
    """

    encryption = traitlets.Enum(["curve"], default_value=None, allow_none=True)
    "The type of encryption to use."

    @override
    def write_connection_file(self, **kwargs: Any) -> None:
        if self.encryption == "curve" and not self.curve_publickey:
            self.curve_publickey, self.curve_secretkey = zmq.curve_keypair()
        if self.curve_publickey:
            self.encryption = "curve"
        return super().write_connection_file(**kwargs)

    async def _connect_socket(self, channel: Channel, /) -> ZMQPollSocket:
        """Create, configure and connect a socket."""
        port = int(getattr(self, f"{channel}_port"))
        assert port
        if channel not in [Channel.iopub, Channel.heartbeat]:
            assert channel not in self._sockets

        def open_socket() -> ZMQPollSocket:
            # Thread: zmq_poll
            port = int(getattr(self, f"{channel}_port"))
            assert port
            if channel is not Channel.iopub:
                assert channel not in self._sockets
            # Open the socket.
            match channel:
                case Channel.heartbeat:
                    socket = self.zmq_poll.socket(zmq.SocketType.REQ)
                case Channel.shell | Channel.control | Channel.stdin:
                    socket = self.zmq_poll.socket(zmq.SocketType.DEALER)
                case Channel.iopub:
                    socket = self.zmq_poll.socket(zmq.SocketType.SUB)
            socket.identity = self.session.bsession
            socket.setsockopt(zmq.SocketOption.LINGER, 500)
            # Encryption.
            if self.curve_secretkey is not None and self.curve_publickey is not None:
                socket.curve_secretkey = self.curve_secretkey
                socket.curve_publickey = self.curve_publickey
                socket.curve_serverkey = self.curve_publickey
            # Bind the socket.
            addr = f"tcp://{self.ip}:{port}" if self.transport == "tcp" else f"ipc://{self.ip}-{port}"
            socket.connect(addr)
            self.log.debug("%s socket connected to %s", channel, addr)
            if channel not in [Channel.iopub, Channel.heartbeat]:
                self._sockets[channel] = socket
            return socket

        return await self.zmq_poll.aexecute(open_socket)

    async def _establish_connection(self, timeout: float | None) -> None:
        # Wait for welcome
        async with self.iopub_subscribe(timeout=timeout):
            pass
        self.log.debug("Getting kernel info to configure session")
        msg = await self.kernel_info()
        adapt_version = int(msg["content"]["protocol_version"].split(".")[0])
        if adapt_version != jupyter_client.protocol_version_info[0]:  # pyright: ignore[reportPrivateImportUsage]
            self.session.adapt_version = adapt_version  # pragma: no cover
        # Send a message on the stdin to ensure the connection is established.
        # This should help prevent input request messages from being silently
        # discarded before the DEALER is properly connected to the SERVER which
        # was observed when running tests that execute code requesting input.
        # An async kernel connection will send a reply to the message.
        with anyio.move_on_after(timeout or 1.0):
            await self.send_message(self.msg(MsgType.kernel_info_request, None, Channel.stdin))
        self.log.debug("Session config complete")

    @override
    def start(self, *, connect_timeout: float | None = None) -> Self:
        """Connect this client to the interface.

        Args:
            connect_timeout: The maximum time to wait for the connection to reply with a welcome message and to configure the session.
                passing `connect_timeout=0` will skip the `_establish_connection` step.
        """
        if not self.shell_port:
            msg = "Connection info has not been set. Tip: consider using the method `subprocess_kernel` or `load_connection_info`."
            raise RuntimeError(msg)
        return super().start(connect_timeout=connect_timeout)

    @override
    async def connection_task(
        self, started: Callable[[], Any], stop: ProtectedPending, *, connect_timeout: float | None = None
    ) -> None:
        def handler(sock, event, channel: Channel, recv=self.session.recv, handle_msg=self.handle_incoming_msg) -> None:
            ident, msg = recv(sock)
            msg["channel"] = channel
            handle_msg(msg, ident)

        connect = self._connect_socket
        async with self.caller:
            with (
                self.zmq_poll as zpoll,
                zpoll.event_handler(await connect(Channel.control), partial(handler, channel=Channel.control)),
                zpoll.event_handler(await connect(Channel.shell), partial(handler, channel=Channel.shell)),
                zpoll.event_handler(await connect(Channel.stdin), partial(handler, channel=Channel.stdin)),
            ):
                if connect_timeout != 0:
                    await self._establish_connection(connect_timeout)
                await super().connection_task(started, stop)
                self._sockets.clear()

    @asynccontextmanager
    async def subprocess_kernel(
        self,
        *,
        connect_timeout: float | None = None,
        heartbeat_interval: float | None = 10.0,
        shutdown_timeout: float | None = 10.0,
        **kwargs,
    ) -> AsyncGenerator[subprocess.Popen]:
        """Start a kernel interface as a subprocess."""
        self.write_connection_file()
        command = make_argv(connection_file=self.connection_file, **kwargs)
        process: subprocess.Popen | None = None
        try:
            # We deliberately use subprocess directly because it is safer in pytest and debugpy.
            async with self.start(connect_timeout=0):
                process = subprocess.Popen(command)  # noqa: ASYNC220
                await self._establish_connection(connect_timeout)
                if heartbeat_interval is not None:
                    hb = self.caller.create_start_stop_task(self._monitor_heartbeat)
                    async with hb.start(interval=heartbeat_interval):
                        yield process
                else:
                    yield process
                await self.shutdown(False).wait(timeout=shutdown_timeout)
                process.wait(timeout=shutdown_timeout)
        finally:
            self.cleanup_connection_file()
            self.cleanup_ipc_files()
            if process and process.returncode is None:
                # Terminate will prevent coverage from writing the necessary files.
                process.terminate()

    async def _monitor_heartbeat(
        self, started: Callable[[], None], stop: ProtectedPending, interval: float = 10.0
    ) -> None:
        reply = "starting"

        def recv(sock: ZMQPollSocket, event: int):
            # Thread: zmq_poll
            nonlocal reply, started
            reply = sock.recv() == b"ping"

        with await self._connect_socket(Channel.heartbeat) as hb:
            ready = create_async_waiter()
            with self.zmq_poll.event_handler(hb, recv, count=(1, ready.wake)):
                hb.send(b"ping")
                await ready
                started()
            with self.zmq_poll.event_handler(hb, recv):
                while not stop.done():
                    reply = ""
                    hb.send(b"ping")
                    try:
                        await stop.wait(timeout=interval)
                    except TimeoutError:
                        if not reply:
                            msg = f"Heartbeat not detected after {interval}s!"
                            raise RuntimeError(msg) from None

    @asynccontextmanager
    @override
    async def iopub_subscribe(
        self, topic=b"", *, timeout: float | None = 10.0
    ) -> AsyncGenerator[SingleAsyncQueue[Message]]:

        def forward_messages(sock: ZMQPollSocket, event: int) -> None:
            msg: Message = self.session.recv(sock)[1]  # pyright: ignore[reportAssignmentType]
            if not ready:
                if msg["header"]["msg_type"] == MsgType.iopub_welcome:
                    ready.set()
            else:
                queue.append(msg)

        queue, ready, scope = SingleAsyncQueue(), create_async_event(), anyio.CancelScope()

        def canceller():
            scope.cancel("ZMQ poll eventloop is stopped!")  # pragma: no cover

        iopub = await self._connect_socket(Channel.iopub)
        with iopub, self.zmq_poll.event_handler(iopub, forward_messages, canceller=canceller), scope:
            iopub.subscribe(topic)
            if timeout is not None:
                self.log.debug("Waiting for welcome message.")
                if await ready.with_(timeout=timeout):
                    self.log.debug("Welcome message received.")
                else:
                    msg = f"Welcome message not received after {timeout:0.1f}s!"
                    raise TimeoutError(msg)
            yield queue

encryption class-attribute instance-attribute 

encryption = traitlets.Enum(['curve'], default_value=None, allow_none=True)

The type of encryption to use.

start 

start(*, connect_timeout: float | None = None) -> Self

Connect this client to the interface.

Parameters:

  • connect_timeout

    (float | None, default: None ) –

    The maximum time to wait for the connection to reply with a welcome message and to configure the session. passing connect_timeout=0 will skip the _establish_connection step.

Source code in src/async_kernel/messaging/zmq.py 
304
305
306
307
308
309
310
311
312
313
314
315
@override
def start(self, *, connect_timeout: float | None = None) -> Self:
    """Connect this client to the interface.

    Args:
        connect_timeout: The maximum time to wait for the connection to reply with a welcome message and to configure the session.
            passing `connect_timeout=0` will skip the `_establish_connection` step.
    """
    if not self.shell_port:
        msg = "Connection info has not been set. Tip: consider using the method `subprocess_kernel` or `load_connection_info`."
        raise RuntimeError(msg)
    return super().start(connect_timeout=connect_timeout)

subprocess_kernel async 

subprocess_kernel(
    *,
    connect_timeout: float | None = None,
    heartbeat_interval: float | None = 10.0,
    shutdown_timeout: float | None = 10.0,
    **kwargs,
) -> AsyncGenerator[Popen]

Start a kernel interface as a subprocess.

Referenced by:
Source code in src/async_kernel/messaging/zmq.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
@asynccontextmanager
async def subprocess_kernel(
    self,
    *,
    connect_timeout: float | None = None,
    heartbeat_interval: float | None = 10.0,
    shutdown_timeout: float | None = 10.0,
    **kwargs,
) -> AsyncGenerator[subprocess.Popen]:
    """Start a kernel interface as a subprocess."""
    self.write_connection_file()
    command = make_argv(connection_file=self.connection_file, **kwargs)
    process: subprocess.Popen | None = None
    try:
        # We deliberately use subprocess directly because it is safer in pytest and debugpy.
        async with self.start(connect_timeout=0):
            process = subprocess.Popen(command)  # noqa: ASYNC220
            await self._establish_connection(connect_timeout)
            if heartbeat_interval is not None:
                hb = self.caller.create_start_stop_task(self._monitor_heartbeat)
                async with hb.start(interval=heartbeat_interval):
                    yield process
            else:
                yield process
            await self.shutdown(False).wait(timeout=shutdown_timeout)
            process.wait(timeout=shutdown_timeout)
    finally:
        self.cleanup_connection_file()
        self.cleanup_ipc_files()
        if process and process.returncode is None:
            # Terminate will prevent coverage from writing the necessary files.
            process.terminate()