Gecko [ primetaxinc.net ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	925 B	-rw-r--r--
__main__.py	422 B	-rw-r--r--
_minimal_curses.py	1.8 KB	-rw-r--r--
_threading_handler.py	2.12 KB	-rw-r--r--
base_eventqueue.py	3.75 KB	-rw-r--r--
commands.py	12.01 KB	-rw-r--r--
completing_reader.py	9.81 KB	-rw-r--r--
console.py	6.71 KB	-rw-r--r--
curses.py	1.21 KB	-rw-r--r--
fancy_termios.py	2.51 KB	-rw-r--r--
historical_reader.py	12.93 KB	-rw-r--r--
input.py	3.69 KB	-rw-r--r--
keymap.py	6.31 KB	-rw-r--r--
main.py	1.89 KB	-rw-r--r--
mypy.ini	870 B	-rw-r--r--
pager.py	5.68 KB	-rw-r--r--
reader.py	27.06 KB	-rw-r--r--
readline.py	19.75 KB	-rw-r--r--
simple_interact.py	5.66 KB	-rw-r--r--
trace.py	433 B	-rw-r--r--
types.py	354 B	-rw-r--r--
unix_console.py	26.09 KB	-rw-r--r--
unix_eventqueue.py	2.46 KB	-rw-r--r--
utils.py	2.39 KB	-rw-r--r--
windows_console.py	21.11 KB	-rw-r--r--
windows_eventqueue.py	991 B	-rw-r--r--

Code Editor : _threading_handler.py

from __future__ import annotations

from dataclasses import dataclass, field
import traceback

TYPE_CHECKING = False
if TYPE_CHECKING:
    from threading import Thread
    from types import TracebackType
    from typing import Protocol

class ExceptHookArgs(Protocol):
        @property
        def exc_type(self) -> type[BaseException]: ...
        @property
        def exc_value(self) -> BaseException | None: ...
        @property
        def exc_traceback(self) -> TracebackType | None: ...
        @property
        def thread(self) -> Thread | None: ...

class ShowExceptions(Protocol):
        def __call__(self) -> int: ...
        def add(self, s: str) -> None: ...

from .reader import Reader

def install_threading_hook(reader: Reader) -> None:
    import threading

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

def show(self) -> int:
            count = 0
            with self.lock:
                if not self.messages:
                    return 0
                reader.restore()
                for tb in self.messages:
                    count += 1
                    if tb:
                        print(tb)
                self.messages.clear()
                reader.scheduled_commands.append("ctrl-c")
                reader.prepare()
            return count

def add(self, s: str) -> None:
            with self.lock:
                self.messages.append(s)

def exception(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" if args.thread else "\n"
            tb = pre + "".join(lines)
            self.add(tb)

def __call__(self) -> int:
            return self.show()

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

${this.title}

Code Editor : _threading_handler.py