run and validate a notebook

checkipnb.py

#!/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)

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

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.

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

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

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

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

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

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

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.