Skip to content

Instantly share code, notes, and snippets.

@minrk
Last active April 9, 2022 16:51
Show Gist options
  • Save minrk/2620876 to your computer and use it in GitHub Desktop.
Save minrk/2620876 to your computer and use it in GitHub Desktop.
run and validate a notebook
#!/usr/bin/env python
"""
simple example script for running notebooks and reporting exceptions.
Usage: `checkipnb.py foo.ipynb [bar.ipynb [...]]`
Each cell is submitted to the kernel, and checked for errors.
"""
import os,sys,time
from Queue import Empty
try:
from IPython.kernel import KernelManager
except ImportError:
from IPython.zmq.blockingkernelmanager import BlockingKernelManager as KernelManager
from IPython.nbformat.current import reads, NotebookNode
def run_notebook(nb):
km = KernelManager()
km.start_kernel(stderr=open(os.devnull, 'w'))
try:
kc = km.client()
except AttributeError:
# 0.13
kc = km
kc.start_channels()
shell = kc.shell_channel
# simple ping:
shell.execute("pass")
shell.get_msg()
cells = 0
failures = 0
for ws in nb.worksheets:
for cell in ws.cells:
if cell.cell_type != 'code':
continue
shell.execute(cell.input)
# wait for finish, maximum 20s
reply = shell.get_msg(timeout=20)['content']
if reply['status'] == 'error':
failures += 1
print "\nFAILURE:"
print cell.input
print '-----'
print "raised:"
print '\n'.join(reply['traceback'])
cells += 1
sys.stdout.write('.')
print
print "ran notebook %s" % nb.metadata.name
print " ran %3i cells" % cells
if failures:
print " %3i cells raised exceptions" % failures
kc.stop_channels()
km.shutdown_kernel()
del km
if __name__ == '__main__':
for ipynb in sys.argv[1:]:
print "running %s" % ipynb
with open(ipynb) as f:
nb = reads(f.read(), 'json')
run_notebook(nb)
@danielballan
Copy link

@hugadams I had the same question, and I discovered this: https://pypi.python.org/pypi/runipy

@GaelVaroquaux
Copy link

Neither the above code, nor runipy, seem to run actually run an IPython notebook as the notebook itself runs it: if I have a "%load" directive in a notebook, and I run all the cells via the interactive menu, the "%load" magic loads the code in a new cell, and this cell is then executed. On the opposite, the various headless runners that I have found just skip the new cell.

It would be really useful to be able to run notebook headless. Currently it is hard to do quality assurance on notebooks.

@zonca
Copy link

zonca commented May 11, 2015

@minrk would it be possible to update this script for IPython 3?

@twiecki
Copy link

twiecki commented Aug 4, 2015

@minrk Second that request. Tried to run but got a:

    shell.execute("pass")
AttributeError: 'ZMQSocketChannel' object has no attribute 'execute'

@kayhan-batmanghelich
Copy link

@minrk I got the same error as @twiecki. AttributeError: 'ZMQSocketChannel' object has no attribute 'execute' am I missing something?

@veroblanchard
Copy link

@twiecki @kayhan-batmanghelich did you fix the issue?

@brunsgaard
Copy link

brunsgaard commented Mar 16, 2017

Our solution

import os
import glob
import time
import sys
import traceback

import nbconvert
import nbformat

ep = nbconvert.preprocessors.ExecutePreprocessor(
    extra_arguments=["--log-level=40"],
    timeout=300,
)

def run_notebook(path):
    path = os.path.abspath(path)
    assert path.endswith('.ipynb')
    nb = nbformat.read(path, as_version=4)
    try:
        ep.preprocess(nb, {'metadata': {'path': os.path.dirname(path)}})
    except Exception as e:
        print("\nException raised while running '{}'\n".format(path))
        traceback.print_exc(file=sys.stdout)
        sys.exit(1)


if __name__ == '__main__':
    print('Running notebooks might take a long time...')
    print('===========================================\n')
    for path in glob.iglob('notebooks/**/*.ipynb', recursive=True):
        root, ext = os.path.splitext(os.path.basename(path))
        if root.endswith('_'):
            continue
        s = time.time()
        sys.stdout.write('Now running ' + path)
        sys.stdout.flush()
        run_notebook(path)
        sys.stdout.write(' -- Finish in {}s\n'.format(int(time.time()-s)))


print('\n\033[92m'
      '==========================='
      ' Notebook testing done '
      '==========================='
      '\033[0m')

@sroet
Copy link

sroet commented Nov 8, 2021

Just incase someone runs into the same issue as me:

>            reply = shell.get_msg(timeout=20)['content']
E               TypeError: 'coroutine' object is not subscriptable

This was solved for me by replacing shell.get_msg with kc.get_shell_msg as mentioned in a note on the 6->7 migration guide.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment