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

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.

source

CaptureShell.run_cell

def run_cell(
    raw_cell, # The code (including IPython code such as %magic functions) to run.
    store_history:bool=False, # If True, the raw and translated cell will be stored in IPython's
    # history. For user code calling back into IPython's machinery, this
    # should be set to False.
    silent:bool=False, # If True, avoid side-effects, such as implicit displayhooks and
    # and logging.  silent=True forces store_history=False.
    shell_futures:bool=True, # If True, the code will share future statements with the interactive
    # shell. It will both be affected by previous __future__ imports, and
    # any __future__ imports in the code will affect the shell. If False,
    # __future__ imports are not shared in either direction.
    cell_id:NoneType=None, # A unique identifier for the cell. This is used in the messaging system
    # to match output with execution requests and for tracking cell execution
    # history across kernel restarts. In notebook contexts, this is typically
    # a UUID generated by the frontend. If None, the kernel may generate an
    # internal identifier or proceed without cell tracking capabilities.
    stdout:bool=True, stderr:bool=True, display:bool=True, timeout:NoneType=None, verbose:bool=False
):

Run a complete IPython cell.

s = CaptureShell(mpl_format='retina')

s.run_cell('a=1');

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

s.run_cell("%%capture c\nprint(1)")
test_eq(s.run_cell("c.stdout").result.result, '1\n')
s.run_cell("r = get_ipython().run_cell('7')")
test_eq(s.run_cell("type(r).__name__").result.result, '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.

o = s.run_cell('# a narration comment\n\n%%capture c2\nprint(2)')
test_eq(o.exception, None)
test_eq(s.run_cell('c2.stdout').result.result, '2\n')

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

o = s.run_cell('print(a)')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: None; err: None; info: <cell: print(a); id: None>,
  'stderr': '',
  'stdout': '1\n'}

o = s.run_cell('from warnings import warn; warn("1")')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: None; err: None; info: <cell: from warnings import warn; warn("1"); id: None>,
  'stderr': '<ipython-input-1-a51443ae013a>:1: UserWarning: 1\n'
            '  from warnings import warn; warn("1")\n',
  'stdout': ''}

o = s.run_cell('1')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result: 1; err: None; info: <cell: 1; id: None>,
  'stderr': '',
  'stdout': ''}

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

{ 'display_objects': [<IPython.utils.capture.RichOutput object>],
  'exception': None,
  'quiet': False,
  'result': result: <IPython.core.display.Markdown object>; err: None; info: <cell: from IPython.display import Markdown,display; print(0); display(Markdown("*2*")); Markdown("*1*"); id: None>,
  'stderr': '',
  'stdout': '0\n'}

o.result.result

o.display_objects[0]

o = s.run_cell('1;')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': True,
  'result': result: 1; err: None; info: <cell: 1;; id: None>,
  'stderr': '',
  'stdout': ''}

o = s.run_cell('import matplotlib.pyplot as plt; plt.plot([1,2,3])')
o

{ 'display_objects': [<IPython.utils.capture.RichOutput object>],
  'exception': None,
  'quiet': False,
  'result': result: [<matplotlib.lines.Line2D object>]; err: None; info: <cell: import matplotlib.pyplot as plt; plt.plot([1,2,3]); id: None>,
  'stderr': '',
  'stdout': ''}

o.result.result[0]

o.display_objects[0]

o = s.run_cell('''
import pandas as pd
pd.DataFrame({'A': [1, 2], 'B': [3, 4]})''')
o

{ 'display_objects': [],
  'exception': None,
  'quiet': False,
  'result': result:    A  B
0  1  3
1  2  4; err: None; info: <cell: 
import pandas as pd
pd.DataFrame({'A': [1, 2], 'B': [3, 4]}); id: None>,
  'stderr': '',
  'stdout': ''}

o.result.result

	A	B
0	1	3
1	2	4

o = s.run_cell('1/0')
o

{ 'display_objects': [],
  'exception': ZeroDivisionError('division by zero'),
  'quiet': False,
  'result': result: None; err: division by zero; info: <cell: 1/0; id: None>,
  'stderr': '',
  'stdout': '\x1b[31m---------------------------------------------------------------------------\x1b[39m\n'
            '\x1b[31mZeroDivisionError\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[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'}

Testing errors caught after exec:

o = s.run_cell('import time; time.sleep(2)', timeout=1)
test_eq(type(o['exception']), TimeoutError)

Testing errors caught before exec:

o = s.run_cell('print(', timeout=1)
test_eq(isinstance(o['exception'], SyntaxError), True)
o = s.run_cell("def foo():\npass")
test_eq(isinstance(o['exception'], IndentationError), True)
o = s.run_cell("if True:\n\tpass\n        pass")
test_eq(isinstance(o['exception'], TabError), True)

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, # Shell command will time out after {timeout} seconds
    verbose:bool=False, # Show stdout/stderr during execution
):

Run code, returning a list of all outputs in Jupyter notebook format

quiet follows IPython’s trailing-semicolon convention for ordinary Python cells. A cell magic’s body is an opaque payload, which may be shell code, markup, or another language. Plain IPython does not let a semicolon at the end of that payload suppress the magic’s result, so CaptureShell only sets quiet for non-cell-magic source.

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;"), [])

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: 1 us, total: 2 us\n',
   'Wall time: 2.62 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

A trailing ; stops the result from being captured:

s.run("1+2;")

[]

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[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[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)
File <ipython-input-1-8ef4b5800722>:1
----> 1 try: f()
      2 except Exception as e:
      3     ex = e

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)
o[0]['text'][:2]

['\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': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ['*2*']},
  'metadata': {},
  'output_type': 'display_data'},
 {'data': {'text/plain': ['<IPython.core.display.Markdown object>'],
   'text/markdown': ['*1*']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

source

CaptureShell.run_async

async def run_async(
    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, # Shell command will time out after {timeout} seconds
    verbose:bool=False, # Show stdout/stderr during execution
):

Call self as a function.

await s.run_async("1+1")

[{'data': {'text/plain': ['2']},
  'metadata': {},
  'output_type': 'execute_result',
  'execution_count': None}]

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
):

Run cell, skipping if not code, and store outputs 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, [])

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
):

Run all cells in nb, stopping at first exception if exc_stop

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!

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
):

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

nbopen / nbrun

Downstream kernels (e.g clikernel) need to run cells from a notebook by cell id, e.g to power %open/%run magics.

source

select_cells

def select_cells(
    nb, # A notebook read with `read_nb`
    msgid:str=None, # Cell id, or unique prefix, to match
    above:bool=False, # Include the matched cell and all cells above it?
    below:bool=False, # Include the matched cell and all cells below it?
    all:bool=False, # Include all code cells (ignores `msgid`)?
    exported:bool=False, # Only cells with `#| export` or `#| exports`?
    skip_noeval:bool=False, # Skip `#| eval: false` and `nbdev_export` cells (like `nbdev-test`)?
):

Select code cells from nb by cell id or unique prefix

This selects code cells by id prefix, optionally including cells above/below the match, or all cells, and optionally filtering to just those with an nbdev export directive:

nb = read_nb(clean)
codes = [c for c in nb.cells if c.cell_type=='code']
c1 = nb.cells[1]
ci = codes.index(c1)
cid = str(c1.id)
test_eq(select_cells(nb, cid), [c1])
test_eq(select_cells(nb, cid[:4]), [c1])
test_eq(select_cells(nb, cid, above=True), codes[:ci+1])
test_eq(select_cells(nb, cid, below=True), codes[ci:])
test_eq(select_cells(nb, all=True), codes)
c1.source = '#| export\n'+c1.source
test_eq(select_cells(nb, all=True, exported=True), [c1])
c1.source = '#| eval: false\n'+c1.source
test_eq(select_cells(nb, all=True, skip_noeval=True), [c for c in codes if c is not c1])
test_eq(select_cells(nb, all=True), codes)
c2 = next(c for c in codes if c is not c1)
c2.source = 'nbdev_export'+'()'
test_eq(select_cells(nb, all=True, skip_noeval=True), [c for c in codes if c not in (c1,c2)])
c2.id = cid[:4]+'0000'
with expect_fail(Exception, cid[:4]): select_cells(nb, cid[:4])
test_eq(select_cells(nb, cid), [c1])

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.

s = CaptureShell()
o = render_text(s.run('print(1); 2'))
test_is('1' in o and '2' in o, True)
test_eq(render_text(s.run('1;')), '')
test_is('ZeroDivisionError' in render_text(s.run('1/0')), True)

Finally, nbopen sets a default notebook for future nbrun calls, and nbrun selects cells and runs them in the order given, printing each cell’s rendered output. The notebook is re-read from disk on every nbrun so file edits are picked up. The default path is stored as a shell attribute (not in user_ns) so a kernel restart clears it naturally, and nbopen raises on a nonexistent path rather than recording it, so a mistyped argument can’t poison later calls that rely on the default.

source

CaptureShell.nbrun

def nbrun(
    *msgids:str, # Cell id prefixes to run, in the order given
    fname:str | pathlib.Path=None, # Notebook path (defaults to last `nbopen`)
    above:bool=False, # Also run all cells above the match?
    below:bool=False, # Also run all cells below the match?
    all:bool=False, # Run all code cells?
    exported:bool=False, # Only cells with `#| export` or `#| exports`?
    skip_noeval:bool=False, # Skip `#| eval: false` and `nbdev_export` cells (like `nbdev-test`)?
):

Run cell(s) from a notebook by id prefix, printing rendered outputs

source

CaptureShell.nbopen

def nbopen(
    fname:str | pathlib.Path
):

Set the default notebook for nbrun

s = CaptureShell()
s.nbopen(clean)
o = render_text(s.run(f'get_ipython().nbrun("{cid}")'))
test_is('1' in o and '2' in o, True)

s = CaptureShell()
s.nbopen(clean)
o = render_text(s.run(f'get_ipython().nbrun("38ab158d", "{cid}")'))
test_is('bold' in o and 'b123d6d0' in o and o.index('bold') < o.index('b123d6d0'), True)
with expect_fail(Exception, 'does_not_exist'): s.nbopen('does_not_exist.ipynb')
test_eq(s._nbrun_fname, Path(clean))

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_cell('from IPython.display import display,Markdown; display(Markdown("x"))')
r.display_objects, r.stdout

([<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
):

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=']

ExecutionResult.__repr__

ExecutionInfo.__repr__

leading_comment_lines

CaptureShell

CaptureShell.run_cell

CaptureShell.load_profile

Cells / run

format_exc

NbResult

CaptureShell.run

CaptureShell.run_async

render_outputs

CaptureShell.cell

find_output

out_exec

out_stream

out_error

NBs

CaptureShell.run_all

CaptureShell.execute

CaptureShell.prettytb

nbopen / nbrun

select_cells

CaptureShell.nbrun

CaptureShell.nbopen

exec_nb

SmartCompleter

CaptureShell.complete

ExecutionResult.repr

ExecutionInfo.repr