# ops
## Imports
``` python
from fastcore.test import test_eq, test_fail
from jupyter_client.session import Session
```
``` python
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
``` python
def iter_timeout(
timeout:NoneType=None, default:int=10
):
```
*Yield remaining seconds until `timeout` expires, using monotonic time*
------------------------------------------------------------------------
source
### parent_id
``` python
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`](https://AnswerDotAI.github.io/conkernelclient/ops.html#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.
``` python
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
``` python
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
``` python
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.
``` python
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.
``` python
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
``` python
def nb_outputs(
msgs
):
```
*Convert iopub `msgs` to nbformat-style output dicts, dropping
non-output messages*
------------------------------------------------------------------------
source
### iopub_streams
``` python
def iopub_streams(
msgs, name:NoneType=None
):
```
*The stream messages in `msgs`, optionally only stream `name`
(‘stdout’/‘stderr’)*
------------------------------------------------------------------------
source
### iopub_msgs
``` python
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)*
``` python
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`](https://AnswerDotAI.github.io/conkernelclient/ops.html#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`](https://AnswerDotAI.github.io/conkernelclient/ops.html#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.
``` python
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
``` python
async def exec_outs(
code, timeout:int=10, kw:VAR_KEYWORD
):
```
*Execute `code` and return just its nbformat-style outputs*
------------------------------------------------------------------------
source
### ConKernelClient.exec_ok
``` python
async def exec_ok(
code, timeout:int=10, kw:VAR_KEYWORD
):
```
*`exec_drain`, asserting the reply status is ok*
------------------------------------------------------------------------
source
### ConKernelClient.exec_drain
``` python
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`).
``` python
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
``` python
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`](https://AnswerDotAI.github.io/conkernelclient/ops.html#evalerror)
if the kernel raises*
------------------------------------------------------------------------
source
### ConKernelClient.user_exprs
``` python
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
``` python
def parse_expr(
s
):
```
*The `literal_eval` of `s` when its form allows, else `s` unchanged*
------------------------------------------------------------------------
source
### EvalError
``` python
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.
``` python
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
``` python
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
``` python
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`](https://AnswerDotAI.github.io/conkernelclient/core.html#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.
``` python
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
``` python
async def input_reply(
value:str
):
```
*Answer the kernel’s pending input_request (from `input()` in an execute
sent with `allow_stdin=True`)*
``` python
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'")
```
``` python
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
``` python
def ctl(
):
```
*Control requests as methods: `await kc.ctl.shutdown(restart=True)`
sends `shutdown_request`*
------------------------------------------------------------------------
source
### ConKernelClient.cmd
``` python
def cmd(
):
```
*Shell requests as methods: `await kc.cmd.history(...)` sends
`history_request`*
------------------------------------------------------------------------
source
### ConKernelClient.dap
``` python
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.
``` python
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')
```
``` python
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
``` python
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.)
``` python
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
``` python
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
``` python
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*
``` python
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`](https://AnswerDotAI.github.io/conkernelclient/ops.html#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.
``` python
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
``` python
if await km.is_alive():
kc.stop_channels()
await km.shutdown_kernel()
```