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

45 statements  

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"""