fix: when appriser fails during an uncaught exception, don't try twice (#73)

* First adjustment

* Improving coverage

* reformat

* Interpret partials

* Potential fix for code scanning alert no. 2: Workflow does not contain permissions

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

---------

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

Author: svaningelgem

.github/workflows/build.yml

@@ -1,4 +1,6 @@
 name: Build and Test
+permissions:
+  contents: read
 
 on:
   push:
@@ -8,12 +10,8 @@ on:
   workflow_dispatch:
 
 jobs:
-  test:
+  lint:
     runs-on: ubuntu-latest
-    strategy:
-      matrix:
-        python-version: ["3.10", "3.11", "3.12", "3.13"]
-      fail-fast: false
 
     steps:
       - uses: actions/checkout@v5
@@ -24,37 +22,58 @@ jobs:
           virtualenvs-create: true
           virtualenvs-in-project: true
 
-      - name: Set up Python ${{ matrix.python-version }}
+      - name: Set up Python 3.10
         uses: actions/setup-python@v6
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: "3.10"
           cache: 'poetry'
 
       - name: Install dependencies
         run: poetry install --no-interaction
 
       - name: Run ruff format check
-        if: always()
         run: poetry run ruff format --check --diff .
 
       - name: Run ruff lint check
-        if: always()
         run: poetry run ruff check --diff .
 
       - name: Run consistency checks between stub & module
-        if: always()
         env:
           PYTHONPATH: "."
         run: poetry run stubtest logprise
 
       - name: Running mypy
-        if: always()
         env:
           PYTHONPATH: "."
         run: poetry run mypy .
 
+  test:
+    needs: lint
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12", "3.13"]
+      fail-fast: false
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Install Poetry
+        uses: snok/install-poetry@v1
+        with:
+          virtualenvs-create: true
+          virtualenvs-in-project: true
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: 'poetry'
+
+      - name: Install dependencies
+        run: poetry install --no-interaction
+
       - name: Run tests with coverage
-        if: always()
         run: poetry run pytest --cov=logprise --cov-report=xml
 
       - name: Upload coverage report
@@ -64,4 +83,4 @@ jobs:
           fail_ci_if_error: true
           files: ./coverage.xml
           verbose: true
-          token: ${{ secrets.CODECOV_TOKEN }}
+          token: ${{ secrets.CODECOV_TOKEN }}

logprise/__init__.py

@@ -2,11 +2,15 @@
 
 import atexit
 import functools
+import inspect
 import logging
 import sys
+import sysconfig
 import threading
+from functools import partial
 from logging import StreamHandler
 from pathlib import Path
+from threading import get_ident
 from typing import TYPE_CHECKING, ClassVar, Final
 
 import apprise.cli
@@ -79,6 +83,9 @@ class Appriser:
     """A wrapper around Apprise to accumulate logs and send notifications."""
 
     _accumulator_id: ClassVar[int | None] = None
+    _exit_via_unhandled_exception: ClassVar[bool] = False
+    _original_excepthook: Callable[[type[BaseException], BaseException, types.TracebackType | None], None] = None
+    _original_threading_excepthook: Callable[[threading.ExceptHookArgs], None] = None
 
     def __init__(
         self,
@@ -104,7 +111,8 @@ def __init__(
         # Initialize everything
         self._load_default_config_paths()
         self._setup_interception_handler()
-        self._setup_exception_hook()
+        self._setup_sys_exception_hook()
+        self._setup_threading_exception_hook()
         self._start_periodic_flush()
         self._setup_at_exit_cleanup()
         self._setup_removal_prevention()
@@ -156,23 +164,110 @@ def new_log_method(self: logging.Logger, *args: object, **kwargs: object) -> Non
         new_log_method._intercepted_by_logprise = True
         logging.Logger._log = new_log_method
 
-    def _setup_exception_hook(self) -> None:
+    def _setup_sys_exception_hook(self) -> None:
         """Set up a hook to capture uncaught exceptions."""
-        original_excepthook = sys.excepthook
-
-        def uncaught_exception(
-            exc_type: type[BaseException], exc_value: BaseException, exc_traceback: types.TracebackType | None
-        ) -> None:
-            # Log the exception
-            logger.opt(exception=(exc_type, exc_value, exc_traceback)).error(
-                f"Uncaught exception: {exc_type.__name__}: {exc_value}"
-            )
-            # Force send the notification immediately for uncaught exceptions
-            self.send_notification()
-            # Call the original excepthook
+        self._original_excepthook = hook = sys.excepthook
+
+        # We want the original one, not go through multiple Appriser objects!
+        while (
+            isinstance(hook, partial)
+            and hook.func.__name__ == self._handle_uncaught_sys_exception.__name__
+            and isinstance(hook.keywords, dict)
+            and "original_excepthook" in hook.keywords
+        ):
+            hook = hook.keywords["original_excepthook"]
+
+        sys.excepthook = partial(self._handle_uncaught_sys_exception, original_excepthook=hook)
+
+    def _setup_threading_exception_hook(self) -> None:
+        """Set up a hook to capture uncaught exceptions."""
+        self._original_threading_excepthook = hook = threading.excepthook
+
+        # We want the original one, not go through multiple Appriser objects!
+        while (
+            isinstance(hook, partial)
+            and hook.func.__name__ == self._handle_uncaught_threading_exception.__name__
+            and isinstance(hook.keywords, dict)
+            and "original_excepthook" in hook.keywords
+        ):
+            hook = hook.keywords["original_excepthook"]
+
+        threading.excepthook = partial(self._handle_uncaught_threading_exception, original_excepthook=hook)
+
+    _STDLIB_BACKPORTS: ClassVar[frozenset[str]] = frozenset(
+        {
+            "exceptiongroup",  # stdlib in 3.11+
+            "importlib_metadata",  # stdlib in 3.8+
+            "importlib_resources",  # stdlib in 3.9+
+            "typing_extensions",  # backport of typing features
+            "tomli",  # tomllib in 3.11+
+        }
+    )
+
+    @staticmethod
+    def _is_method_in_stdlib(method: Callable) -> bool:
+        module = inspect.getmodule(method)
+        if not module:
+            return False
+
+        top_level = method.__module__.split(".", maxsplit=1)[0]
+
+        if top_level in sys.stdlib_module_names or top_level in Appriser._STDLIB_BACKPORTS:
+            return True
+
+        if not hasattr(module, "__file__") or not module.__file__:
+            return True
+
+        module_path = Path(module.__file__).resolve().absolute()
+
+        if "site-packages" in module_path.parts or "dist-packages" in module_path.parts:
+            return False
+
+        all_paths = sysconfig.get_paths()
+        for check_this in ["stdlib", "platstdlib"]:
+            path = Path(all_paths[check_this]).resolve().absolute()
+            if module_path.is_relative_to(path):
+                return True
+
+        return False
+
+    def _handle_uncaught_sys_exception(
+        self,
+        exc_type: type[BaseException],
+        exc_value: BaseException,
+        exc_traceback: types.TracebackType | None,
+        original_excepthook: Callable[[type[BaseException], BaseException, types.TracebackType | None], None],
+    ) -> None:
+        """Handle uncaught exceptions by logging and sending notifications."""
+        logger.opt(exception=(exc_type, exc_value, exc_traceback)).error(
+            f"Uncaught exception: {exc_type.__name__}: {exc_value}"
+        )
+
+        Appriser._exit_via_unhandled_exception = True
+
+        self.send_notification()
+
+        if not self._is_method_in_stdlib(original_excepthook):
             original_excepthook(exc_type, exc_value, exc_traceback)
 
-        sys.excepthook = uncaught_exception
+    def _handle_uncaught_threading_exception(
+        self,
+        args: threading.ExceptHookArgs,
+        /,
+        original_excepthook: Callable[[threading.ExceptHookArgs], None],
+    ) -> None:
+        """Handle uncaught exceptions by logging and sending notifications."""
+        logger.opt(exception=(args.exc_type, args.exc_value, args.exc_traceback)).error(
+            f"Uncaught exception in thread {args.thread.name if args.thread else get_ident()}:"
+            f" {args.exc_type.__name__}: {args.exc_value}"
+        )
+
+        Appriser._exit_via_unhandled_exception = True
+
+        self.send_notification()
+
+        if not self._is_method_in_stdlib(original_excepthook):
+            original_excepthook(args)
 
     @property
     def flush_interval(self) -> int | float:
@@ -232,7 +327,12 @@ def stop_periodic_flush(self) -> None:
     def cleanup(self) -> None:
         """Clean up resources and send any pending notifications."""
         self.stop_periodic_flush()
-        self.send_notification()
+
+        if not Appriser._exit_via_unhandled_exception:
+            self.send_notification()
+
+        sys.excepthook = self._original_excepthook
+        threading.excepthook = self._original_threading_excepthook
 
     @property
     def notification_level(self) -> int:
@@ -263,15 +363,19 @@ def send_notification(
     ) -> None:
         """Send a single notification with all accumulated logs."""
         if not self.buffer:
+            logger.debug("No logs to send")
             return
 
         # Format the buffered logs into a single message
         message = "".join(self.buffer).replace("\r", "")
 
-        if self.apprise_obj and self.apprise_obj.notify(
-            title=title, notify_type=notify_type, body=message, body_format=body_format
-        ):
-            self.buffer.clear()  # Clear the buffer after sending
+        try:
+            if message and self.apprise_obj.notify(
+                title=title, notify_type=notify_type, body=message, body_format=body_format
+            ):
+                self.buffer.clear()  # Clear the buffer after sending
+        except BaseException as e:
+            logger.warning(f"Failed to send notification: {e}")
 
 
 appriser = Appriser()

tests/conftest.py

@@ -1,9 +1,12 @@
+import sys
+import threading
+from collections.abc import Generator
 from typing import Any
 
 import pytest
-from apprise import Apprise, NotifyBase, NotifyType
+from apprise import NotifyBase, NotifyType
 
-import logprise
+from logprise import Appriser, logger
 
 
 class NoOpNotifier(NotifyBase):
@@ -15,37 +18,55 @@ class NoOpNotifier(NotifyBase):
     # Define protocol(s) this notification supports
     protocol = ("noop", "dummy")
 
+    calls: list[dict]
+
     def __init__(self, **kwargs):
         super().__init__(secure=False, **kwargs)
+        self.calls = []
 
     def url(self, privacy: bool = False, *args: Any, **kwargs: Any) -> str:
         return "noop://"
 
     def send(self, body: str, title: str = "", notify_type: NotifyType = NotifyType.INFO, **kwargs: Any) -> bool:
         # Simply return True to simulate successful notification
+        self.calls.append({"title": title, "body": body})
         return True
 
 
-@pytest.fixture()
-def noop() -> NoOpNotifier:
-    return NoOpNotifier()
+@pytest.fixture
+def apprise_noop() -> Generator[tuple[Appriser, NoOpNotifier], Any, None]:
+    a = Appriser()
+    noop = NoOpNotifier()
+    a.add(noop)
+    try:
+        yield a, noop
+    finally:
+        a.cleanup()
 
 
 @pytest.fixture(autouse=True)
-def notify_mock(monkeypatch):
-    """Only mock the notify method of Apprise"""
-    calls = []
+def reset_appriser_object() -> Generator[None, None, None]:
+    try:
+        yield
+    finally:
+        Appriser._exit_via_unhandled_exception = False
 
-    def mock_notify(self, title, body, *_, **__):
-        calls.append({"title": title, "body": body})
-        return True
 
-    monkeypatch.setattr(Apprise, "notify", mock_notify)
-    return calls
+@pytest.fixture(autouse=True)
+def save_restore_excepthooks() -> Generator[None, None, None]:
+    original_excepthook = sys.excepthook
+    original_threading_excepthook = threading.excepthook
+    try:
+        yield
+    finally:
+        sys.excepthook = original_excepthook
+        threading.excepthook = original_threading_excepthook
 
 
 @pytest.fixture(autouse=True)
 def silence_logger():
-    logprise.logger.remove()  # Silence any output
-    yield
-    logprise.logger.remove()  # And restore any handlers we added
+    logger.remove()  # Silence any output
+    try:
+        yield
+    finally:
+        logger.remove()  # And restore any handlers we added