shell

A shell for running notebook code without a notebook server

from fastcore.ansi import ansi2html
from IPython.display import HTML
from fastcore.test import *
from base64 import b64decode
from io import BytesIO
from PIL import Image
import tempfile
from contextlib import contextmanager

---

source

ExecutionResult.__repr__

def __repr__():

Return repr(self).

---

source

ExecutionInfo.__repr__

def __repr__():

Return repr(self).

---

source

leading_comment_lines

def leading_comment_lines(
    lines
):

Input cleanup transform: drop leading blank/comment lines when the first code line is a cell magic, which IPython otherwise misparses as a line magic

s = InteractiveShell()

---

source

CaptureShell

def CaptureShell(
    path:str | pathlib.Path=None, mpl_format:str='retina', history:bool=False, timeout:Optional=None,
    profile:bool=False
):

An enhanced, interactive shell for Python.

Every CaptureShell owns a private event loop, run forever by a daemon thread created at construction. All cell execution happens on that loop, through exactly one crossing point: run submits the coroutine with run_coroutine_threadsafe and blocks on the returned future. Nothing else ever touches the loop, so loop identity can’t be confused whatever context the caller is in – a sync CLI, an async server, or another kernel – and a timeout bounds the wait even when a cell blocks the loop in sync code. A timed-out or wedged shell may be unusable afterwards (its loop thread might still be inside the stuck cell): discard it and make a fresh one rather than reusing it.

Captured execution is factored into three pieces. _captured wraps a run with output capture and stdin blocking. _run_res packages the result and captured streams as the AttrDict that rendering uses. The run call itself comes in two forms sharing those pieces: _run_async awaits fastcore.nbio.run_cell and is what run submits to the shell’s loop, while sync _run goes through IPython’s own run_cell entry on the calling thread, for constructor-time setup lines.

---

source

CaptureShell.load_profile

def load_profile(
    name:str='default'
):

Load profile name’s extensions and run its startup files (output suppressed), as ipykernel does

profile=True makes a CaptureShell behave like ipykernel at startup: it reads the IPython profile (honoring IPYTHONDIR), applies shell config traits from ipython_config.py and ipython_kernel_config.py, loads their InteractiveShellApp.extensions, and runs the profile’s startup files. (exec_lines/exec_files are not run.)

To demo profile loading we need a temporary IPython profile:

@contextmanager
def _tmp_profile(cfg='', kcfg='', startup=''):
    with tempfile.TemporaryDirectory() as td, modified_env(IPYTHONDIR=td):
        pd = Path(td)/'profile_default'
        (pd/'startup').mkdir(parents=True)
        if cfg: (pd/'ipython_config.py').write_text(cfg)
        if kcfg: (pd/'ipython_kernel_config.py').write_text(kcfg)
        if startup: (pd/'startup'/'00-start.py').write_text(startup)
        yield td

Startup files in the profile’s startup directory are run, so their variables appear in the shell’s namespace:

with _tmp_profile(startup="a=7\n"): test_eq(CaptureShell(profile=True).user_ns['a'], 7)

Extensions and startup files often print or display things as they load; load_profile suppresses that output so it can’t leak into test runs or captured results (failures still surface as warnings):

with _tmp_profile(startup="b=1\nprint('starting up')\n"):
    with capture_output() as cap: sp = CaptureShell(profile=True)
test_eq((cap.stdout, sp.user_ns['b']), ('', 1))

Shell config traits from the profile’s config files are applied too:

with _tmp_profile(kcfg="c.InteractiveShell.ast_node_interactivity='all'\n"):
    test_eq(CaptureShell(profile=True).ast_node_interactivity, 'all')

With the default history=False, the shell’s history database lives in memory, so creating shells never writes a history.sqlite into the profile directory:

with _tmp_profile() as td:
    CaptureShell(profile=True)
    test_eq((Path(td)/'profile_default'/'history.sqlite').exists(), False)

Cells / run

---

source

format_exc

def format_exc(
    e
):

Format exception e as a string list

---

source

NbResult

def NbResult(
    *args, **kwargs
):

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

---

source

CaptureShell.run

def run(
    code:str, # Python/IPython code to run
    stdout:bool=True, # Capture stdout and save as output?
    stderr:bool=True, # Capture stderr and save as output?
    timeout:Optional=None, # Seconds before the run times out (None: `self.timeout`)
    verbose:bool=False, # Show stdout/stderr during execution
    grace:float=5, # Seconds to wait for cancellation after a timeout
):

Run code on the shell’s loop, returning a list of all outputs in Jupyter notebook format

Captured runs pass store_history=True, so IPython’s own trailing-; suppression applies, and the check is tokenizer-based: a comment after the ; still suppresses. A cell magic’s payload is unaffected, since its transformed source ends in the run_cell_magic(...) call rather than a semicolon. That is the right behavior for payloads, which may be shell code, markup, or another language where a trailing ; means nothing. A suppressed result is never populated, so no execute_result output appears. Streams are unaffected either way:

cs = CaptureShell(mpl_format=None)
cs.register_magic_function(lambda line, cell: 'visible', 'cell', 'visible')
res = cs.run("%%visible\npayload;")
test_eq(res[0]['data']['text/plain'], ["'visible'"])
test_eq(cs.run("1+1;"), [])
test_eq(cs.run("print('kept'); 2;"), [dict(output_type='stream', name='stdout', text=['kept\n'])])

s = CaptureShell()

s.run("print(1)")

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']}]

Code can include magics and ! shell commands:

o = s.run("%time 1+1")
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['CPU times: user 1e+03 ns, sys: 0 ns, total: 1e+03 ns\n',
   'Wall time: 1.91 us\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

The result of the last successful execution is stored in result:

s.result

2

A trailing ; stops the result from being captured:

s.run("1+2;")

[]

Code running inside a cell can itself call run_cell (the %%capture magic does exactly that). Such nested calls behave exactly as in plain IPython. They return an ExecutionResult, and their output flows to the enclosing capture context, so %%capture works:

s.run("%%capture c\nprint(1)")
test_eq(s.user_ns['c'].stdout, '1\n')
s.run("r = get_ipython().run_cell('7')")
test_eq(s.user_ns['r'].__class__.__name__, 'ExecutionResult')

IPython requires a cell magic to be the very first line of a cell: even a comment above it makes the %%foo line parse as a line magic named %foo, which fails with a confusing “Line magic function %%foo not found” error. CaptureShell smooths both edges: leading blank and comment lines above a cell magic are dropped (extending IPython’s own leading_empty_lines cleanup), and when real code precedes the magic (which can’t be auto-fixed), the error message explains the actual problem.

s.run('# a narration comment\n\n%%capture c2\nprint(2)')
test_eq(s.exc, None)
test_eq(s.user_ns['c2'].stdout, '2\n')

s.run('a=1\n%%capture c3\nprint(3)')
assert 'must start the cell' in str(s.exc)

o = s.run("1/0")
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mZeroDivisionError\x1b[39m                         Traceback (most recent call last)\n',
   '\x1b[36mFile \x1b[39m\x1b[32m<ipython-input-1-9e1622b385b6>:1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[30;43m1\x1b[39;49m\x1b[30;43m/\x1b[39;49m\x1b[30;43m0\x1b[39;49m\n',
   '\n',
   '\x1b[31mZeroDivisionError\x1b[39m: division by zero\n']},
 {'ename': 'ZeroDivisionError',
  'evalue': 'division by zero',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/aai-ws/.venv/lib/python3.13/site-packages/IPython/core/interactiveshell.py", line 3748, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n',
   '  File "<ipython-input-1-9e1622b385b6>", line 1, in <module>\n    1/0\n    ~^~\n',
   'ZeroDivisionError: division by zero\n']}]

This is how IPython formats exceptions internally:

from IPython.core.ultratb import VerboseTB

with warnings.catch_warnings():
    warnings.filterwarnings("ignore", category=DeprecationWarning)
    formatter = VerboseTB(color_scheme='Linux')

try: f()
except Exception as e:
    ex = e
    print(formatter.text(type(e), e, e.__traceback__))

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[58], line 4
      1 try: f()
      2 except Exception as e:
      3     ex = e
----> 4     print(formatter.text(type(e), e, e.__traceback__))

NameError: name 'f' is not defined

s.run("import time; time.sleep(0.1); print('no timeout')", timeout=1)

[{'name': 'stdout', 'output_type': 'stream', 'text': ['no timeout\n']}]

o = s.run("import time; time.sleep(1.1)", timeout=1)
test_eq(o[0]['ename'], 'TimeoutError')
s.exc

['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
 '\x1b[31mTimeoutError\x1b[39m                              Traceback (most recent call last)\n']

o1 = s.run('from IPython.display import Markdown,display; print(0); print(1); display(Markdown("*2*")); Markdown("*1*")')
o1

[{'name': 'stdout', 'output_type': 'stream', 'text': ['0\n', '1\n']},
 {'data': {'text/plain': ['Markdown(*2*)'], 'text/markdown': ['*2*']},
  'metadata': {},
  'output_type': 'display_data'},
 {'data': {'text/plain': ['Markdown(*1*)'], 'text/markdown': ['*1*']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

oaw = render_text(s.run("import asyncio\nawait asyncio.sleep(0.01)\n7"))
test_is('7' in oaw, True)

A captured run reports its exception structurally, packaged as an error output, so IPython’s own traceback print would be a colored duplicate in the captured stream. showtraceback therefore does nothing.

---

source

CaptureShell.showsyntaxerror

def showsyntaxerror(
    *args, **kwargs
):

Like showtraceback

---

source

CaptureShell.showtraceback

def showtraceback(
    *args, **kwargs
):

A captured run reports its exception structurally; IPython’s printed traceback would be a colored duplicate in the stream

An error cell yields exactly one output, the structural error, with no traceback text leaking into the stream:

s = CaptureShell()
test_eq([x['output_type'] for x in s.run('1/0')], ['error'])

---

source

render_outputs

def render_outputs(
    outputs, ansi_renderer:function=_strip, include_imgs:bool=True, pygments:bool=False, md_tfm:function=noop,
    html_tfm:function=noop
):

Call self as a function.

HTML(render_outputs(o))

---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
Cell In[1], line 1
----> 1 import time; time.sleep(1.1)

File <ipython-input-1-289b30000b65>:7, in run_cell.<locals>.handler(*args)
      5 if not timeout: timeout = self.timeout
      6 if timeout:
----> 7     def handler(*args): raise TimeoutError()
      8     signal.signal(signal.SIGALRM, handler)
      9     signal.alarm(timeout)

TimeoutError: 

We can use ansi2html to convert from ANSI to HTML for color rendering. You need some css styles for the colors to render properly. Jupyter already has these built in so it’s not neccessary here, but if you plan on using this in another web app you will need to ensure that css styling is included.

HTML(render_outputs(o, ansi2html))

---------------------------------------------------------------------------
TimeoutError                              Traceback (most recent call last)
Cell In[1], line 1
----> 1 import time; time.sleep(1.1)

File <ipython-input-1-289b30000b65>:7, in run_cell.<locals>.handler(*args)
      5 if not timeout: timeout = self.timeout
      6 if timeout:
----> 7     def handler(*args): raise TimeoutError()
      8     signal.signal(signal.SIGALRM, handler)
      9     signal.alarm(timeout)

TimeoutError: 

Images and matplotlib figures are captured:

res = s.run('''import matplotlib.pyplot as plt
plt.figure(figsize=(2,1))
plt.plot([1,2,4]);''')

HTML(render_outputs(res))

If an exception is raised then the exception type, object, and stacktrace are stored in exc:

o = s.run('raise Exception("Oops")')
o

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mException\x1b[39m                                 Traceback (most recent call last)\n',
   '\x1b[36mCell\x1b[39m\x1b[36m \x1b[39m\x1b[32mIn[1]\x1b[39m\x1b[32m, line 1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[38;5;28;01mraise\x1b[39;00m \x1b[38;5;167;01mException\x1b[39;00m(\x1b[33m"\x1b[39m\x1b[33mOops\x1b[39m\x1b[33m"\x1b[39m)\n',
   '\n',
   '\x1b[31mException\x1b[39m: Oops\n']},
 {'ename': 'Exception',
  'evalue': 'Oops',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/aai-ws/.venv/lib/python3.13/site-packages/IPython/core/interactiveshell.py", line 3748, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n',
   '  File "<ipython-input-1-01648acb07bd>", line 1, in <module>\n    raise Exception("Oops")\n',
   'Exception: Oops\n']}]

s.exc

Exception('Oops')

---

source

CaptureShell.cell

def cell(
    cell, stdout:bool=True, stderr:bool=True, verbose:bool=False, timeout:NoneType=None
):

Run cell, skipping if not code, and store outputs (and the execution count) back in cell

clean = Path('../tests/clean.ipynb')
nb = read_nb(clean)
c = nb.cells[1]
c

{ 'cell_type': 'code',
  'execution_count': None,
  'id': 'b123d6d0',
  'idx_': 1,
  'metadata': {},
  'outputs': [],
  'source': 'print(1)\n2'}

s.cell(c)
c.outputs

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

Re-running a cell that no longer produces output clears its stale outputs:

c2 = read_nb(clean).cells[1]
s.cell(c2)
assert c2.outputs
c2.source = 'x = 1'
s.cell(c2)
test_eq(c2.outputs, [])

cell also records how long the run took, in the metadata.execution dict that Jupyter’s record_timing option uses: ISO 8601 timestamps under the standard message-key names, plus a total key holding elapsed seconds for programmatic use (run_all’s postproc receives the cell right after these are stamped). In an nbdev project nbdev-clean strips this metadata, so it stays out of committed notebooks unless allowed with allowed_cell_metadata_keys.

timing = c.metadata.execution
assert 0 <= timing['total'] < 10
test_eq(set(timing), {'iopub.execute_input','shell.execute_reply','total'})
list(timing)

---

source

find_output

def find_output(
    outp, # Output from `run`
    ot:str='execute_result', # Output_type to find
):

Find first output of type ot in CaptureShell.run output

find_output(c.outputs)['data']

{'text/plain': ['2']}

find_output(c.outputs, 'stream')['text']

['1\n']

---

source

out_exec

def out_exec(
    outp
):

Get data from execution result in outp.

out_exec(c.outputs)

'2'

---

source

out_stream

def out_stream(
    outp
):

Get text from stream in outp.

out_stream(c.outputs)

'1'

---

source

out_error

def out_error(
    outp
):

Get traceback from error in outp.

NBs

---

source

CaptureShell.run_all

def run_all(
    nb, # A notebook read with `nbclient` or `read_nb`
    exc_stop:bool=False, # Stop on exceptions?
    preproc:callable=_false, # Called before each cell is executed
    postproc:callable=_false, # Called after each cell is executed
    inject_code:str | None=None, # Code to inject into a cell
    inject_idx:int=0, # Cell to replace with `inject_code`
    verbose:bool=False, # Show stdout/stderr during execution
    cell_timeout:int=None, # Seconds before each cell times out (None: no limit)
    grace:float=5, # Seconds to wait for cancellation before giving up
):

Run all cells in nb, stopping at first exception if exc_stop; tasks a run leaves behind are cancelled, with survivors in self.leaks

nb.cells[2].outputs

[]

s.run_all(nb)
nb.cells[2].outputs

[{'data': {'text/plain': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ["This is *bold*. Here's a [link](https://www.fast.ai)."]},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

With exc_stop=False (the default), execution continues after exceptions, and exception details are stored into the appropriate cell’s output:

nb.cells[-1].source

'raise Exception("Oopsie!")'

nb.cells[-1].outputs

[{'name': 'stdout',
  'output_type': 'stream',
  'text': ['\x1b[31m---------------------------------------------------------------------------\x1b[39m\n',
   '\x1b[31mException\x1b[39m                                 Traceback (most recent call last)\n',
   '\x1b[36mCell\x1b[39m\x1b[36m \x1b[39m\x1b[32mIn[1]\x1b[39m\x1b[32m, line 1\x1b[39m\n',
   '\x1b[32m----> \x1b[39m\x1b[32m1\x1b[39m \x1b[38;5;28;01mraise\x1b[39;00m \x1b[38;5;167;01mException\x1b[39;00m(\x1b[33m"\x1b[39m\x1b[33mOopsie!\x1b[39m\x1b[33m"\x1b[39m)\n',
   '\n',
   '\x1b[31mException\x1b[39m: Oopsie!\n']},
 {'ename': 'Exception',
  'evalue': 'Oopsie!',
  'output_type': 'error',
  'traceback': ['Traceback (most recent call last):\n',
   '  File "/Users/jhoward/aai-ws/.venv/lib/python3.13/site-packages/IPython/core/interactiveshell.py", line 3748, in run_code\n    exec(code_obj, self.user_global_ns, self.user_ns)\n    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n',
   '  File "<ipython-input-1-1c97c1d317ab>", line 1, in <module>\n    raise Exception("Oopsie!")\n',
   'Exception: Oopsie!\n']}]

With exc_stop=True, exceptions in a cell are raised and no further processing occurs:

try: s.run_all(nb, exc_stop=True)
except Exception as e: print(f"got exception: {e}")

got exception: Oopsie!

A hung cell can’t hang the caller: cell_timeout bounds each cell at the submission boundary, recording a TimeoutError (including the stacks of the loop’s pending tasks) as the cell’s output. When the run ends, tasks it started but never finished are cancelled, with any that refuse to die left in leaks:

s2 = CaptureShell(mpl_format=None)
nb2 = new_nb([mk_cell('import asyncio'), mk_cell('t = asyncio.ensure_future(asyncio.sleep(60))'), mk_cell('await asyncio.sleep(60)')])
s2.run_all(nb2, cell_timeout=1, grace=1)
test_eq(nb2.cells[-1].outputs[0]['ename'], 'TimeoutError')
test_eq(s2.leaks, [])
test_eq(s2.user_ns['t'].cancelled(), True)

Cancellation needs the loop: a cell blocking in sync code pins the loop, so nothing can be delivered to it. The timeout still returns on schedule – the caller is never hung – but the shell’s thread stays stuck inside the cell, the stuck tasks land in leaks, and the shell should be discarded:

s3 = CaptureShell(mpl_format=None)
nb3 = new_nb([mk_cell('import time'), mk_cell('time.sleep(4)')])
s3.run_all(nb3, cell_timeout=1, grace=0.2)
test_eq(nb3.cells[-1].outputs[0]['ename'], 'TimeoutError')
assert s3.leaks

We can pass a function to preproc to have it run on every cell. It can modify the cell as needed. If the function returns True, then that cell will not be executed. For instance, to skip the cell which raises an exception:

nb = read_nb(clean)
s.run_all(nb, preproc=lambda c: 'raise' in c.source)

This cell will contain no output, since it was skipped.

nb.cells[-1].outputs

[]

nb.cells[1].outputs

[{'name': 'stdout', 'output_type': 'stream', 'text': ['1\n']},
 {'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

You can also pass a function to postproc to modify a cell after it is executed.

---

source

CaptureShell.execute

def execute(
    src:str | pathlib.Path, # Notebook path to read from
    dest:str | None=None, # Notebook path to write to
    exc_stop:bool=False, # Stop on exceptions?
    preproc:callable=_false, # Called before each cell is executed
    postproc:callable=_false, # Called after each cell is executed
    inject_code:str | None=None, # Code to inject into a cell
    inject_path:str | pathlib.Path | None=None, # Path to file containing code to inject into a cell
    inject_idx:int=0, # Cell to replace with `inject_code`
    verbose:bool=False, # Show stdout/stderr during execution
    cell_timeout:int=None, # Seconds before each cell times out (None: no limit)
):

Execute notebook from src and save with outputs to `dest

This is a shortcut for the combination of read_nb, CaptureShell.run_all, and write_nb.

s = CaptureShell()
try:
    s.execute(clean, 'tmp.ipynb')
    print(read_nb('tmp.ipynb').cells[1].outputs)
finally: Path('tmp.ipynb').unlink()

[{'name': 'stdout', 'output_type': 'stream', 'text': '1\n'}, {'data': {'text/plain': '2'}, 'execution_count': None, 'metadata': {}, 'output_type': 'execute_result'}]

p = Path.home()/'git'/'fastcore'/'nbs'
n = p/'03a_parallel.ipynb'

---

source

CaptureShell.prettytb

def prettytb(
    fname:str | pathlib.Path=None, # filename to print alongside the traceback
):

Show a pretty traceback for notebooks, optionally printing fname.

If an error occurs while running a notebook, you can retrieve a pretty version of the error with the prettytb method:

s = CaptureShell()
try: s.execute('../tests/error.ipynb', exc_stop=True)
except: print(s.prettytb())

AssertionError in ../tests/error.ipynb:
===========================================================================

While Executing Cell #2:
Traceback (most recent call last):
  File "<ipython-input-1-5c812627fe60>", line 2, in <module>
    try: s.execute('../tests/error.ipynb', exc_stop=True)
         ~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<ipython-input-1-4b062e10fd76>", line 19, in execute
    self.run_all(nb, exc_stop=exc_stop, preproc=preproc, postproc=postproc,
    ~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
        inject_code=inject_code, inject_idx=inject_idx, verbose=verbose)
        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<ipython-input-1-54c4c86f5c38>", line 21, in run_all
    if self.exc and exc_stop: raise self.exc from None
                              ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/jhoward/aai-ws/.venv/lib/python3.13/site-packages/IPython/core/interactiveshell.py", line 3748, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<ipython-input-1-b968a57a586e>", line 3, in <module>
    foo()
    ~~~^^
  File "/Users/jhoward/aai-ws/execnb/tests/err.py", line 2, in foo
    assert 13 == 98
           ^^^^^^^^
AssertionError

render_text

Downstream kernels (e.g clikernel) print each executed cell’s outputs as plain text, so nbio’s render_text is re-exported here for them. Running notebook cells by id in the current kernel is aidialog’s %nbrun magic, built on fastcore.nbio.select_cells and fastcore.nbio.run_cell.

run_text names the composition those kernels use: run code and return what a notebook’s output area would show, as concise text. One visible output renders bare (markdown-preferred), several arrive in render_text’s tagged form, and nothing visible is ''.

---

source

CaptureShell.run_text

def run_text(
    code, stdout:bool=True, # Capture stdout and save as output?
    stderr:bool=True, # Capture stderr and save as output?
    timeout:Optional=None, # Seconds before the run times out (None: `self.timeout`)
    verbose:bool=False, # Show stdout/stderr during execution
    grace:float=5, # Seconds to wait for cancellation after a timeout
):

Run code, returning its outputs rendered as concise text: what a notebook’s output area shows

s = CaptureShell()
o = s.run_text('print(1); 2')
test_is('1' in o and '2' in o, True)
test_eq(s.run_text('1;'), '')
test_eq(s.run_text("class M:\n    def _repr_markdown_(self): return '**hi**'\nM()"), '**hi**')
test_is('ZeroDivisionError' in s.run_text('1/0'), True)
o

render_outputs is HTML-oriented. Downstream kernels print each executed cell’s result to stdout instead, so we use fastcore’s render_text, which renders outputs to concise text.

def is_sublist(sub, lst): return any(lst[i:i+len(sub)] == sub for i in range(len(lst)-len(sub)+1))

s2 = CaptureShell()
r = s2.run('from IPython.display import display,Markdown; display(Markdown("x"))')
test_eq(r[0]['output_type'], 'display_data')
r

([<IPython.utils.capture.RichOutput>], '')

If you pass inject_code to CaptureShell.execute or CaptureShell.run_all, the source of nb.cells[inject_idx] will be replaced with inject_code. By default, the first cell is replaced. For instance consider this notebook:

nb = read_nb('../tests/params.ipynb')
for c in nb.cells: print('- ',c.source)

-  a=1
-  print(a)

We can replace the first cell with a=2 by passing that as inject_code, and the notebook will run with that change:

nb = read_nb('../tests/params.ipynb')
s.run_all(nb, inject_code="a=2")
list(nb.cells)

[{'cell_type': 'code',
  'execution_count': None,
  'id': 'a63ce885',
  'metadata': {'time_run': '2026-01-04T20:52:46.502210+00:00'},
  'outputs': [],
  'source': 'a=2',
  'idx_': 0},
 {'cell_type': 'code',
  'execution_count': None,
  'id': 'ea528db5',
  'metadata': {'time_run': '2026-01-04T20:52:46.506607+00:00'},
  'outputs': [{'name': 'stdout', 'output_type': 'stream', 'text': ['2\n']}],
  'source': 'print(a)',
  'idx_': 1}]

This can be used with CaptureShell.execute to parameterise runs of models in notebooks. Place any defaults for configuration code needed in the first cell, and then when running execute pass in new parameters as needed in inject_code. To replace only some of the defaults, leave an empty cell as the second cell, and inject code using inject_idx=1 to replace the empty second cell with code that overrides some of the defaults set in the first cell. When using execute you can pass inject_path instead of inject_code to read the injected code from a file.

---

source

exec_nb

def exec_nb(
    src:str, # Notebook path to read from
    dest:str='', # Notebook path to write to
    exc_stop:bool=False, # Stop on exceptions?
    inject_code:str=None, # Code to inject into a cell
    inject_path:str=None, # Path to file containing code to inject into a cell
    inject_idx:int=0, # Cell to replace with `inject_code`
    verbose:bool=False, # Show stdout/stderr during execution
    cell_timeout:int=None, # Seconds before each cell times out (None: no limit)
):

Execute notebook from src and save with outputs to dest

This is the command-line version of CaptureShell.execute. Run exec_nb -h from the command line to see how to pass arguments. If you don’t pass dest then the output notebook won’t be saved; this is mainly useful for running tests.

---

source

SmartCompleter

def SmartCompleter(
    shell, # a pointer to the ipython shell itself.  This is needed
    # because this completer knows about magic functions, and those can
    # only be accessed via the ipython instance.
    namespace:NoneType=None, # an optional dict where completions are performed.
    jedi:bool=False
):

Extension of the completer class with IPython-specific features

cc = SmartCompleter(get_ipython())

def test_set(a,b): return test_eq(set(a), set(b))

class _f:
    def __init__(self): self.bar,self.baz,self.room = 0,0,0

foo = _f()

assert set(cc("b")).issuperset(['bool', 'bytes'])
test_set(cc("foo.b"), ['bar', 'baz'])
test_set(cc("x=1; x = foo.b"), ['bar', 'baz'])
test_set(cc("ab"), ['abs'])
test_set(cc("b = ab"), ['abs'])
test_set(cc(""), [])
test_set(cc("foo."), ['bar', 'baz', 'room'])
test_set(cc("nonexistent.b"), [])
test_set(cc("foo.nonexistent.b"), [])
assert set(cc("import ab")).issuperset(['abc'])
test_set(cc("from abc import AB"), ['ABC', 'ABCMeta'])

s = CaptureShell()
cc = SmartCompleter(s)
s.run('''def captures(pat, s, n, **kw):
    return 1''')
cc('captures(')

['n=', 'pat=', 's=']

---

source

CaptureShell.complete

def complete(
    c
): # The actual text that was completed.

Return the completed text and a list of completions.

s = CaptureShell()
s.run('a=1')
s.complete('a.b')

['bit_count', 'bit_length']

s.run('import re')
s.complete('re.compile(')

['flags=', 'pattern=']