1"""Exceptions for nbclient."""
2from __future__ import annotations
3
4from typing import Any
5
6from nbformat import NotebookNode
7
8
9class CellControlSignal(Exception): # noqa
10 """
11 A custom exception used to indicate that the exception is used for cell
12 control actions (not the best model, but it's needed to cover existing
13 behavior without major refactors).
14 """
15
16 pass
17
18
19class CellTimeoutError(TimeoutError, CellControlSignal):
20 """
21 A custom exception to capture when a cell has timed out during execution.
22 """
23
24 @classmethod
25 def error_from_timeout_and_cell(
26 cls, msg: str, timeout: int, cell: NotebookNode
27 ) -> CellTimeoutError:
28 """Create an error from a timeout on a cell."""
29 if cell and cell.source:
30 src_by_lines = cell.source.strip().split("\n")
31 src = (
32 cell.source
33 if len(src_by_lines) < 11
34 else f"{src_by_lines[:5]}\n...\n{src_by_lines[-5:]}"
35 )
36 else:
37 src = "Cell contents not found."
38 return cls(timeout_err_msg.format(timeout=timeout, msg=msg, cell_contents=src))
39
40
41class DeadKernelError(RuntimeError):
42 """A dead kernel error."""
43
44 pass
45
46
47class CellExecutionComplete(CellControlSignal):
48 """
49 Used as a control signal for cell execution across execute_cell and
50 process_message function calls. Raised when all execution requests
51 are completed and no further messages are expected from the kernel
52 over zeromq channels.
53 """
54
55 pass
56
57
58class CellExecutionError(CellControlSignal):
59 """
60 Custom exception to propagate exceptions that are raised during
61 notebook execution to the caller. This is mostly useful when
62 using nbconvert as a library, since it allows to deal with
63 failures gracefully.
64 """
65
66 def __init__(self, traceback: str, ename: str, evalue: str) -> None:
67 """Initialize the error."""
68 super().__init__(traceback)
69 self.traceback = traceback
70 self.ename = ename
71 self.evalue = evalue
72
73 def __reduce__(self) -> tuple[Any]:
74 """Reduce implementation."""
75 return type(self), (self.traceback, self.ename, self.evalue) # type:ignore[return-value]
76
77 def __str__(self) -> str:
78 """Str repr."""
79 if self.traceback:
80 return self.traceback
81 else:
82 return f"{self.ename}: {self.evalue}"
83
84 @classmethod
85 def from_cell_and_msg(cls, cell: NotebookNode, msg: dict[str, Any]) -> CellExecutionError:
86 """Instantiate from a code cell object and a message contents
87 (message is either execute_reply or error)
88 """
89
90 # collect stream outputs for our error message
91 stream_outputs: list[str] = []
92 for output in cell.outputs:
93 if output["output_type"] == "stream":
94 stream_outputs.append(
95 stream_output_msg.format(name=output["name"], text=output["text"].rstrip())
96 )
97 if stream_outputs:
98 # add blank line before, trailing separator
99 # if there is any stream output to display
100 stream_outputs.insert(0, "")
101 stream_outputs.append("------------------")
102 stream_output: str = "\n".join(stream_outputs)
103
104 tb = "\n".join(msg.get("traceback", []) or [])
105 return cls(
106 exec_err_msg.format(
107 cell=cell,
108 stream_output=stream_output,
109 traceback=tb,
110 ),
111 ename=msg.get("ename", "<Error>"),
112 evalue=msg.get("evalue", ""),
113 )
114
115
116stream_output_msg: str = """\
117----- {name} -----
118{text}"""
119
120exec_err_msg: str = """\
121An error occurred while executing the following cell:
122------------------
123{cell.source}
124------------------
125{stream_output}
126
127{traceback}
128"""
129
130
131timeout_err_msg: str = """\
132A cell timed out while it was being executed, after {timeout} seconds.
133The message was: {msg}.
134Here is a preview of the cell contents:
135-------------------
136{cell_contents}
137-------------------
138"""