cpython/Lib/_pyrepl/_threading_handler.py at 3.13 · python/cpython · GitHub

Name: cpython/Lib/_pyrepl/_threading_handler.py at 3.13 · python/cpython · GitHub
Rating: 4.6 (6305 reviews)
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
from __future__ importannotations

fromdataclassesimportdataclass, field
importtraceback


TYPE_CHECKING=False
ifTYPE_CHECKING:
fromthreadingimportThread
fromtypesimportTracebackType
fromtypingimportProtocol

classExceptHookArgs(Protocol):
@property
defexc_type(self) ->type[BaseException]: ...
@property
defexc_value(self) ->BaseException|None: ...
@property
defexc_traceback(self) ->TracebackType|None: ...
@property
defthread(self) ->Thread|None: ...

classShowExceptions(Protocol):
def__call__(self) ->int: ...
defadd(self, s: str) ->None: ...

from .readerimportReader


definstall_threading_hook(reader: Reader) ->None:
importthreading

@dataclass
classExceptHookHandler:
lock: threading.Lock=field(default_factory=threading.Lock)
messages: list[str] =field(default_factory=list)

defshow(self) ->int:
count=0
withself.lock:
ifnotself.messages:
return0
reader.restore()
fortbinself.messages:
count+=1
iftb:
print(tb)
self.messages.clear()
reader.scheduled_commands.append("ctrl-c")
reader.prepare()
returncount

defadd(self, s: str) ->None:
withself.lock:
self.messages.append(s)

defexception(self, args: ExceptHookArgs) ->None:
lines=traceback.format_exception(
args.exc_type,
args.exc_value,
args.exc_traceback,
colorize=reader.can_colorize,
 ) # type: ignore[call-overload]
pre=f"\nException in {args.thread.name}:\n"ifargs.threadelse"\n"
tb=pre+"".join(lines)
self.add(tb)

def__call__(self) ->int:
returnself.show()


handler=ExceptHookHandler()
reader.threading_hook=handler
threading.excepthook=handler.exception