Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/nbclient/exceptions.py: 49%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""Exceptions for nbclient."""
2from __future__ import annotations
4from typing import Any
6from nbformat import NotebookNode
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 """
16 pass
19class CellTimeoutError(TimeoutError, CellControlSignal):
20 """
21 A custom exception to capture when a cell has timed out during execution.
22 """
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))
41class DeadKernelError(RuntimeError):
42 """A dead kernel error."""
44 pass
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 """
55 pass
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 """
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
73 def __reduce__(self) -> tuple[Any]:
74 """Reduce implementation."""
75 return type(self), (self.traceback, self.ename, self.evalue) # type:ignore[return-value]
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}"
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 """
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)
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 )
116stream_output_msg: str = """\
117----- {name} -----
118{text}"""
120exec_err_msg: str = """\
121An error occurred while executing the following cell:
122------------------
123{cell.source}
124------------------
125{stream_output}
127{traceback}
128"""
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"""