1from __future__ import annotations 2 3from dataclasses import dataclass, field 4import traceback 5 6 7TYPE_CHECKING = False 8if TYPE_CHECKING: 9 from threading import Thread 10 from types import TracebackType 11 from typing import Protocol 12 13 class ExceptHookArgs(Protocol): 14 @property 15 def exc_type(self) -> type[BaseException]: ... 16 @property 17 def exc_value(self) -> BaseException | None: ... 18 @property 19 def exc_traceback(self) -> TracebackType | None: ... 20 @property 21 def thread(self) -> Thread | None: ... 22 23 class ShowExceptions(Protocol): 24 def __call__(self) -> int: ... 25 def add(self, s: str) -> None: ... 26 27 from .reader import Reader 28 29 30def install_threading_hook(reader: Reader) -> None: 31 import threading 32 33 @dataclass 34 class ExceptHookHandler: 35 lock: threading.Lock = field(default_factory=threading.Lock) 36 messages: list[str] = field(default_factory=list) 37 38 def show(self) -> int: 39 count = 0 40 with self.lock: 41 if not self.messages: 42 return 0 43 reader.restore() 44 for tb in self.messages: 45 count += 1 46 if tb: 47 print(tb) 48 self.messages.clear() 49 reader.scheduled_commands.append("ctrl-c") 50 reader.prepare() 51 return count 52 53 def add(self, s: str) -> None: 54 with self.lock: 55 self.messages.append(s) 56 57 def exception(self, args: ExceptHookArgs) -> None: 58 lines = traceback.format_exception( 59 args.exc_type, 60 args.exc_value, 61 args.exc_traceback, 62 colorize=reader.can_colorize, 63 ) # type: ignore[call-overload] 64 pre = f"\nException in {args.thread.name}:\n" if args.thread else "\n" 65 tb = pre + "".join(lines) 66 self.add(tb) 67 68 def __call__(self) -> int: 69 return self.show() 70 71 72 handler = ExceptHookHandler() 73 reader.threading_hook = handler 74 threading.excepthook = handler.exception 75