-
Notifications
You must be signed in to change notification settings - Fork 315
/
execute.py
85 lines (72 loc) · 2.88 KB
/
execute.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from nbconvert.preprocessors import ExecutePreprocessor, CellExecutionError
from traitlets import Bool, List, Integer, validate, TraitError
from textwrap import dedent
from . import NbGraderPreprocessor
from nbconvert.exporters.exporter import ResourcesDict
from nbformat.notebooknode import NotebookNode
from typing import Any, Optional, Tuple
class UnresponsiveKernelError(Exception):
pass
class Execute(NbGraderPreprocessor, ExecutePreprocessor):
timeout = Integer(
30,
help=ExecutePreprocessor.timeout.help,
allow_none=True,
).tag(config=True)
interrupt_on_timeout = Bool(
True,
help=ExecutePreprocessor.interrupt_on_timeout.help
).tag(config=True)
allow_errors = Bool(
True,
help=dedent(
"""
When a cell execution results in an error, continue executing the rest of
the notebook. If False, the thrown nbclient exception would break aspects of
output rendering.
"""
),
)
raise_on_iopub_timeout = Bool(
True,
help=ExecutePreprocessor.raise_on_iopub_timeout.help
).tag(config=True)
error_on_timeout = {
"ename": "CellTimeoutError",
"evalue": "",
"traceback": ["ERROR: No reply from kernel"]
}
extra_arguments = List([], help=dedent(
"""
A list of extra arguments to pass to the kernel. For python kernels,
this defaults to ``--HistoryManager.hist_file=:memory:``. For other
kernels this is just an empty list.
""")
).tag(config=True)
execute_retries = Integer(0, help=dedent(
"""
The number of times to try re-executing the notebook before throwing
an error. Generally, this shouldn't need to be set, but might be useful
for CI environments when tests are flaky.
""")
).tag(config=True)
def on_cell_executed(self, **kwargs):
cell = kwargs['cell']
reply = kwargs['execute_reply']
if reply['content']['status'] == 'error':
error_recorded = False
for output in cell.outputs:
if output.output_type == 'error':
error_recorded = True
if not error_recorded:
# Occurs when
# IPython.core.interactiveshell.InteractiveShell.showtraceback
# = lambda *args, **kwargs : None
error_output = NotebookNode(output_type='error')
error_output.ename = reply['content']['ename']
error_output.evalue = reply['content']['evalue']
error_output.traceback = reply['content']['traceback']
if error_output.traceback == []:
error_output.traceback = ["ERROR: An error occurred while"
" showtraceback was disabled"]
cell.outputs.append(error_output)