Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/nbclient/exceptions.py: 47%

1"""Exceptions for nbclient.""" 

2from typing import Dict, List 

3 

4from nbformat import NotebookNode 

5 

6 

7class CellControlSignal(Exception): # noqa 

8 """ 

9 A custom exception used to indicate that the exception is used for cell 

10 control actions (not the best model, but it's needed to cover existing 

11 behavior without major refactors). 

12 """ 

13 

14 pass 

15 

16 

17class CellTimeoutError(TimeoutError, CellControlSignal): 

18 """ 

19 A custom exception to capture when a cell has timed out during execution. 

20 """ 

21 

22 @classmethod 

23 def error_from_timeout_and_cell( 

24 cls, msg: str, timeout: int, cell: NotebookNode 

25 ) -> "CellTimeoutError": 

26 """Create an error from a timeout on a cell.""" 

27 if cell and cell.source: 

28 src_by_lines = cell.source.strip().split("\n") 

29 src = ( 

30 cell.source 

31 if len(src_by_lines) < 11 

32 else f"{src_by_lines[:5]}\n...\n{src_by_lines[-5:]}" 

33 ) 

34 else: 

35 src = "Cell contents not found." 

36 return cls(timeout_err_msg.format(timeout=timeout, msg=msg, cell_contents=src)) 

37 

38 

39class DeadKernelError(RuntimeError): 

40 """A dead kernel error.""" 

41 

42 pass 

43 

44 

45class CellExecutionComplete(CellControlSignal): 

46 """ 

47 Used as a control signal for cell execution across execute_cell and 

48 process_message function calls. Raised when all execution requests 

49 are completed and no further messages are expected from the kernel 

50 over zeromq channels. 

51 """ 

52 

53 pass 

54 

55 

56class CellExecutionError(CellControlSignal): 

57 """ 

58 Custom exception to propagate exceptions that are raised during 

59 notebook execution to the caller. This is mostly useful when 

60 using nbconvert as a library, since it allows to deal with 

61 failures gracefully. 

62 """ 

63 

64 def __init__(self, traceback: str, ename: str, evalue: str) -> None: 

65 """Initialize the error.""" 

66 super().__init__(traceback) 

67 self.traceback = traceback 

68 self.ename = ename 

69 self.evalue = evalue 

70 

71 def __reduce__(self) -> tuple: 

72 """Reduce implementation.""" 

73 return type(self), (self.traceback, self.ename, self.evalue) 

74 

75 def __str__(self) -> str: 

76 """Str repr.""" 

77 if self.traceback: 

78 return self.traceback 

79 else: 

80 return f"{self.ename}: {self.evalue}" 

81 

82 @classmethod 

83 def from_cell_and_msg(cls, cell: NotebookNode, msg: Dict) -> "CellExecutionError": 

84 """Instantiate from a code cell object and a message contents 

85 (message is either execute_reply or error) 

86 """ 

87 

88 # collect stream outputs for our error message 

89 stream_outputs: List[str] = [] 

90 for output in cell.outputs: 

91 if output["output_type"] == "stream": 

92 stream_outputs.append( 

93 stream_output_msg.format(name=output["name"], text=output["text"].rstrip()) 

94 ) 

95 if stream_outputs: 

96 # add blank line before, trailing separator 

97 # if there is any stream output to display 

98 stream_outputs.insert(0, "") 

99 stream_outputs.append("------------------") 

100 stream_output: str = "\n".join(stream_outputs) 

101 

102 tb = '\n'.join(msg.get('traceback', []) or []) 

103 return cls( 

104 exec_err_msg.format( 

105 cell=cell, 

106 stream_output=stream_output, 

107 traceback=tb, 

108 ), 

109 ename=msg.get('ename', '<Error>'), 

110 evalue=msg.get('evalue', ''), 

111 ) 

112 

113 

114stream_output_msg: str = """\ 

115----- {name} ----- 

116{text}""" 

117 

118exec_err_msg: str = """\ 

119An error occurred while executing the following cell: 

120------------------ 

121{cell.source} 

122------------------ 

123{stream_output} 

124 

125{traceback} 

126""" 

127 

128 

129timeout_err_msg: str = """\ 

130A cell timed out while it was being executed, after {timeout} seconds. 

131The message was: {msg}. 

132Here is a preview of the cell contents: 

133------------------- 

134{cell_contents} 

135------------------- 

136"""