ops

Standard async operations on a running kernel, for ConKernelClient

Imports

from fastcore.test import test_eq, test_fail
from jupyter_client.session import Session

km = ConKernelManager(session=Session(key=b'x'))
await km.start_kernel()
kc = await km.client().start_channels()
await kc.is_alive()

Message basics

---

source

iter_timeout

def iter_timeout(
    timeout:NoneType=None, default:int=10
):

Yield remaining seconds until timeout expires, using monotonic time

---

source

parent_id

def parent_id(
    msg
):

The msg_id of the request this msg responds to, or None

Every reply carries its request’s msg_id in the parent header; all routing below keys on it. iter_timeout drives every wait loop in this module: it yields the time still available, so a loop body can pass a shrinking timeout to each blocking call.

r = await kc.execute('1+1', reply=True)
test_eq(parent_id(r), r['parent_header']['msg_id'])
assert parent_id({}) is None

Iopub collection

---

source

ConKernelClient.iopub_drain

async def iopub_drain(
    msg_id, timeout:int=10
):

Collect iopub messages parented to msg_id until its idle status arrives. Other requests’ messages are discarded: one drain at a time

---

source

ConKernelClient.iopub_flush

async def iopub_flush(
    timeout:float=0.1
):

Discard all pending iopub messages, e.g. leftovers from fire-and-forget executes

iopub_flush is the counterpart for when you don’t want collection: clear the backlog before starting a push-based consumer or a fresh interaction.

kc.execute('print("junk")')
await kc.iopub_flush()
try:
    await kc.get_iopub_msg(timeout=0.1)
    assert False, 'iopub should be empty after flush'
except Empty: pass

Collecting a request’s output means draining iopub until the kernel publishes the idle status for that request - that is the protocol’s “this request is done publishing” signal, far more reliable than sweeping whatever is pending after a fixed delay.

mid = kc.execute('print("hi"); 42')
msgs = await kc.iopub_drain(mid)
test_eq([m['msg_type'] for m in msgs if m['msg_type']!='status'], ['execute_input', 'stream', 'execute_result'])

Outputs

---

source

nb_outputs

def nb_outputs(
    msgs
):

Convert iopub msgs to nbformat-style output dicts, dropping non-output messages

---

source

iopub_streams

def iopub_streams(
    msgs, name:NoneType=None
):

The stream messages in msgs, optionally only stream name (‘stdout’/‘stderr’)

---

source

iopub_msgs

def iopub_msgs(
    msgs, msg_type:NoneType=None
):

Filter iopub msgs by msg_type - a single type, or a collection of types (all messages if None)

test_eq([m['msg_type'] for m in iopub_msgs(msgs, 'execute_result')], ['execute_result'])
test_eq([m['msg_type'] for m in iopub_msgs(msgs, output_types)], ['stream', 'execute_result'])
test_eq(iopub_streams(msgs)[0]['content']['text'], 'hi\n')
test_eq(iopub_streams(msgs, 'stderr'), [])

Raw-message filters for when the protocol wrapping is wanted (parent ids, per-message metadata) - nb_outputs above is the lossy convenience, these are the lossless ones.

Iopub messages carry protocol wrapping (headers, status chatter, execute_input echoes) that consumers rarely want. nb_outputs reduces them to the same output dicts a notebook file stores, so downstream code (e.g. renderers expecting nbformat shapes) can consume them directly. This mirrors nbformat.v4.output_from_msg without the dependency, minus schema validation.

outs = nb_outputs(msgs)
test_eq(outs[0], dict(output_type='stream', name='stdout', text='hi\n'))
test_eq(outs[1]['data']['text/plain'], '42')

Execute composites

---

source

ConKernelClient.exec_outs

async def exec_outs(
    code, timeout:int=10, kw:VAR_KEYWORD
):

Execute code and return just its nbformat-style outputs

---

source

ConKernelClient.exec_ok

async def exec_ok(
    code, timeout:int=10, kw:VAR_KEYWORD
):

exec_drain, asserting the reply status is ok

---

source

ConKernelClient.exec_drain

async def exec_drain(
    code, timeout:int=10, kw:VAR_KEYWORD
):

Execute code; return (reply, outputs) where outputs are the request’s iopub messages

The composites cover the three common shapes: full protocol detail (exec_drain), tests that just need success (exec_ok), and consumers that only want outputs (exec_outs).

reply, outputs = await kc.exec_ok('x = 3; x')
test_eq(reply['content']['status'], 'ok')
test_eq((await kc.exec_outs('print(x*2)')), [dict(output_type='stream', name='stdout', text='6\n')])
reply, _ = await kc.exec_drain('1/0')
test_eq(reply['content']['ename'], 'ZeroDivisionError')

Expression values

---

source

ConKernelClient.eval_expr

async def eval_expr(
    expr:str, code:str='', timeout:int=10, kw:VAR_KEYWORD
):

Evaluate expr in the kernel (optionally after running code) and return its value: parsed via literal_eval when its repr allows, else the repr string. Raises EvalError if the kernel raises

---

source

ConKernelClient.user_exprs

async def user_exprs(
    exprs:dict, code:str='', timeout:int=10, kw:VAR_KEYWORD
):

Run code and evaluate each of the exprs expressions in the same round trip; returns the reply content (statuses and mimebundles unparsed)

---

source

parse_expr

def parse_expr(
    s
):

The literal_eval of s when its form allows, else s unchanged

---

source

EvalError

def EvalError(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

An eval_expr expression raised in the kernel

The user_expressions round trip: one execute carries the expression, the kernel evaluates it after the (empty) cell and returns its repr inside the execute_reply - no iopub involved, so it works cleanly alongside streaming output. Values whose repr is not literal_eval-able come back as that repr string. This is the general core of solveit’s richer eval RPC.

await kc.exec_ok("v = [1, 'a', {'b': 2}]")
test_eq(await kc.eval_expr('v'), [1, 'a', {'b': 2}])
test_eq(await kc.eval_expr('w * 2', code='w = 21'), 42)
cts = await kc.user_exprs({'a': 'v[0]', 'b': 'len(v)'})
test_eq(cts['status'], 'ok')
test_eq(parse_expr(cts['user_expressions']['a']['data']['text/plain']), 1)
test_eq(parse_expr(cts['user_expressions']['b']['data']['text/plain']), 3)
test_eq(await kc.eval_expr('sum(v[2].values())'), 2)
r = await kc.eval_expr('print')
assert isinstance(r, str) and 'print' in r, r
try:
    await kc.eval_expr('nope_undefined')
    assert False, 'expected EvalError'
except EvalError as e: assert 'NameError' in str(e)

Generic requests

---

source

ConKernelClient.control_request

async def control_request(
    msg_type, timeout:int=10, content:VAR_KEYWORD
):

Send a request on the control channel and await its matching reply

---

source

ConKernelClient.shell_request

def shell_request(
    msg_type, timeout:int=10, reply:bool=True, buffers:NoneType=None,
    msg_id:NoneType=None, # Override the auto-generated message id
    subshell_id:NoneType=None, # Route via this subshell (header field)
    content:VAR_KEYWORD
):

Send an arbitrary shell request, routed through the reply reader like execute. Returns a coroutine for the reply, or just the msg_id when reply=False (for fire-and-forget types like comm_open that never get replies)

ConKernelClient’s background reader consumes every shell message, so a bare get_shell_msg would race it and lose. Any non-execute shell request (kernel_info_request, complete_request, …) must therefore register with the same routing table execute uses - that is what shell_request does. Pass reply=False for fire-and-forget message types the kernel never answers (comm_open, comm_msg, comm_close), and buffers for binary payloads. Control-channel replies have no reader competing for them, so control_request matches by parent id directly. Note control requests should not be issued concurrently with each other.

r = await kc.shell_request('kernel_info_request')
test_eq(r['header']['msg_type'], 'kernel_info_reply')
r = await kc.control_request('kernel_info_request')
test_eq(r['header']['msg_type'], 'kernel_info_reply')

Read the input_request from the stdin channel with get_stdin_msg, then answer it; the blocked execute’s reply follows.

---

source

ConKernelClient.input_reply

async def input_reply(
    value:str
):

Answer the kernel’s pending input_request (from input() in an execute sent with allow_stdin=True)

c = kc.execute("x = input('name? ')", reply=True, timeout=10, allow_stdin=True)
msg = await kc.get_stdin_msg(timeout=10)
test_eq(msg['header']['msg_type'], 'input_request')
await kc.input_reply('Ada')
test_eq((await c)['content']['status'], 'ok')
test_eq((await kc.exec_outs('x'))[0]['data']['text/plain'], "'Ada'")

code = """def _t(comm, msg):
    global nbuf; nbuf = len(msg.get('buffers', []))
get_ipython().kernel.comm_manager.register_target('t', _t)
"""
await kc.exec_ok(code)
mid = kc.shell_request('comm_open', reply=False, comm_id='c1', target_name='t', data={}, buffers=[b'12345'])
assert isinstance(mid, str)
await kc.exec_ok('import time; time.sleep(0.2)')
test_eq((await kc.exec_outs('nbuf'))[0]['data']['text/plain'], '1')

Proxies

---

source

ConKernelClient.ctl

def ctl(
    
):

Control requests as methods: await kc.ctl.shutdown(restart=True) sends shutdown_request

---

source

ConKernelClient.cmd

def cmd(
    
):

Shell requests as methods: await kc.cmd.history(...) sends history_request

---

source

ConKernelClient.dap

def dap(
    
):

DAP debug requests as methods. Stateful (seq counter), so cached per client, unlike cmd/ctl

dap speaks the Debug Adapter Protocol over debug_request control messages. A trailing underscore escapes Python keywords (kc.dap.continue_(...) sends the DAP continue command); full=True returns the whole reply message instead of just its content.

The proxies make one-off protocol requests read like methods without hand-assembling messages. They are stateless, so no caching is needed.

r = await kc.cmd.complete(code='pri', cursor_pos=3)
assert 'print' in r['content']['matches']
r = await kc.cmd.is_complete(code='def f():')
test_eq(r['content']['status'], 'incomplete')

r = await kc.shell_request('kernel_info_request', msg_id='fixed-id-1')
test_eq(parent_id(r), 'fixed-id-1')
r = await kc.dap.debugInfo()
test_eq(r['command'], 'debugInfo')
assert r['success']
r2 = await kc.dap.debugInfo(full=True)
test_eq(parent_id(r2), parent_id(r2))  # full message shape
assert r2['header']['msg_type'] == 'debug_reply'

Interrupt

---

source

ConKernelClient.interrupt

async def interrupt(
    timeout:int=5
):

Interrupt the running request via interrupt_request on control; returns the reply

The interrupted execute returns an error reply with ename KeyboardInterrupt, and the kernel stays usable. (Solveit’s gateway historically followed the interrupt with an empty execute to wake a parked recv; the Session.send patch in core addresses that wake-up at the transport level, so it is not repeated here.)

task = asyncio.ensure_future(kc.execute('import time; time.sleep(30)', reply=True, timeout=15))
await asyncio.sleep(0.5)
r = await kc.interrupt()
test_eq(r['header']['msg_type'], 'interrupt_reply')
reply = await task
test_eq(reply['content']['ename'], 'KeyboardInterrupt')
test_eq((await kc.exec_outs('40+2'))[0]['data']['text/plain'], '42')

Lifecycle

---

source

run_kernel

def run_kernel(
    kernel_name:str='python3', # Kernelspec name to launch
    manager_cls:MetaHasTraits=ConKernelManager, # Manager class, e.g. a subclass customizing launch
    kwargs:VAR_KEYWORD
):

Start a kernel, yield (km, kc) with channels running, and always shut both down

---

source

reconnect

async def reconnect(
    km, kc:NoneType=None
):

Restart km’s kernel (fresh process, state discarded) and return a newly connected client, stopping kc if given

async with run_kernel() as (km3, kc3):
    await kc3.exec_ok('x = 42')
    pid1 = km3.provisioner.pid
    kc3 = await reconnect(km3, kc3)
    assert km3.provisioner.pid != pid1
    reply, _ = await kc3.exec_drain('x')
    test_eq(reply['content']['ename'], 'NameError')
    test_eq((await kc3.exec_outs('1+1'))[0]['data']['text/plain'], '2')

After a restart the old client’s channels and routed-reply reader are stale, so reconnect hands back a fresh client bound to the restarted process. State is gone: redo any imports/setup.

The context manager guarantees teardown even when a test body raises, which is what keeps a big protocol test suite from leaking kernel processes.

async with run_kernel() as (km2, kc2):
    test_eq((await kc2.exec_outs('1+1'))[0]['data']['text/plain'], '2')
    pid = km2.provisioner.pid
assert not await km2.is_alive()

Cleanup

if await km.is_alive():
    kc.stop_channels()
    await km.shutdown_kernel()