refactor: add wrapper for catching on handled exceptions

Author: Daniel-Brai

.github/workflows/ci.yml

@@ -19,16 +19,16 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
 
       - name: Install uv
-        uses: astral-sh/setup-uv@v5
+        uses: astral-sh/setup-uv@v7
         with:
           enable-cache: true
           cache-dependency-glob: uv.lock
 
       - name: Set up Python
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v6
         with:
           python-version-file: ".python-version"
 

CONTRIBUTING.md

@@ -22,6 +22,10 @@ cd sqlnotify
 1. **Install dependencies**
 
 ```bash
+# Using make
+make install-dev
+
+# Use uv
 uv sync --all-groups --extra all
 ```
 
@@ -51,6 +55,16 @@ make test
 uv run pytest
 ```
 
+### Run tests with coverage
+
+```bash
+# Using make
+make test_cov
+
+# Or directly with pytest
+uv run pytest --cov=sqlnotify --cov-report=html
+```
+
 ## Code Style
 
 SQLNotify follows Python best practices and uses automated tools to maintain code quality:

Makefile

@@ -1,7 +1,12 @@
 UV := uv
-CODECOV_CLI := codecovcli
 
-.PHONY: test build publish release
+.PHONY: test test_cov build publish release install-dev
+
+
+install-dev:
+	@echo "Installing development dependencies..."
+	@$(UV) sync --all-groups --extra all
+	@echo "Development dependencies installed"
 
 
 build:
@@ -25,3 +30,9 @@ test:
 	@echo "Running all tests for sqlnotify..."
 	@docker compose run --remove-orphans sqlnotify bash -c "$(UV) run pytest"
 	@echo "All tests completed"
+
+
+test_cov:
+	@echo "Running all tests with coverage for sqlnotify..."
+	@docker compose run --remove-orphans sqlnotify bash -c "$(UV) run pytest --cov=sqlnotify --cov-report=html"
+	@echo "All tests completed with coverage report generated"

README.md

@@ -323,7 +323,7 @@ await notifier.anotify(
 # notifier.notify(User, Operation.UPDATE, payload={"id": 123})
 ```
 
-**Note**: The `notify`/`anotify` methods validate payload size and can use overflow tables for large payloads. If `use_overflow_table=True`, payloads exceeding NOTIFY limit are automatically stored in the overflow table, and only an overflow ID is sent through the notification channel.
+**Note**: The `notify`/`anotify` methods validate payload size and can use overflow tables for large payloads. If `use_overflow_table=True`, payloads exceeding SQLNotify limit are automatically stored in the overflow table, and only an overflow ID is sent through the notification channel.
 
 ### Model Change Detection
 
@@ -548,7 +548,7 @@ notifier.start()  # Synchronous start
 2. **Trigger Creation** - The dialect creates database-specific triggers on your tables
 3. **Change Detection** - When data changes, triggers fire and call notification functions
 4. **NOTIFY** - Functions use database-native notification mechanisms (e.g., PostgreSQL's NOTIFY)
-5. **LISTEN** - SQLNotify maintains a dedicated connection listening for notifications
+5. **LISTEN** - SQLNotify maintains a dedicated connection listening for notifications or polls for changes (SQLite)
 6. **Event Distribution** - Incoming notifications are routed to subscribed callbacks
 7. **Callback Execution** - Your subscriber functions are called with change events
 

pyproject.toml

@@ -44,14 +44,13 @@ dependencies = ["sqlalchemy>=2.0.0", "asyncpg>=0.31.0"]
 [dependency-groups]
 test = [
     "pre-commit>=4.5.1",
-    "coverage>=7.13.3",
+    "coverage>=7.13.4",
     "pytest>=9.0.2",
     "pytest-cov>=7.0.0",
     "pytest-xdist>=3.8.0",
     "pytest-asyncio>=1.3.0",
-    "psycopg[binary]>=3.3.2",
-    "sqlmodel>=0.0.25",
-    "httpx>=0.28.0",
+    "psycopg[binary]==3.3.2",
+    "sqlmodel==0.0.33",
 ]
 
 

src/sqlnotify/constants.py

@@ -1,8 +1,8 @@
 PACKAGE_NAME = "sqlnotify"
 
-MAX_SQLNOTIFY_PAYLOAD_BYTES = 7999  # PostgreSQL NOTIFY payload limit (8000 - 1 for terminator)
+MAX_SQLNOTIFY_PAYLOAD_BYTES = 7999  # SQLNotify payload limit (8000 - 1 for terminator)
 
-MAX_SQLNOTIFY_IDENTIFER_BYTES = 63  # PostgreSQL identifier limit (63 bytes)
+MAX_SQLNOTIFY_IDENTIFER_BYTES = 63  # SQLNotify identifier limit (63 bytes)
 
 MAX_SQLNOTIFY_EXTRA_COLUMNS = 5  # Limit extra columns to help stay within payload size limit, but this is not a hard limit since column data size can vary greatly
 

src/sqlnotify/notifiers/notifier.py

@@ -10,7 +10,7 @@
 from ..exceptions import SQLNotifyConfigurationError
 from ..logger import get_logger
 from ..types import ChangeEvent, Operation
-from ..utils import extract_database_url, strip_database_query_params
+from ..utils import extract_database_url, strip_database_query_params, wrap_unhandled_error
 from ..watcher import Watcher
 from .base import BaseNotifier
 
@@ -135,6 +135,7 @@ def dialect_name(self) -> str:
     def is_running(self) -> bool:
         return self._running
 
+    @wrap_unhandled_error(lambda self: self._logger)
     def start(self) -> None:
         """
         Start watching for database changes synchronously.
@@ -151,6 +152,7 @@ def start(self) -> None:
 
         self._start_sync()
 
+    @wrap_unhandled_error(lambda self: self._logger)
     async def astart(self) -> None:
         """
         Start watching for database changes asynchronously.
@@ -203,6 +205,7 @@ def _start_sync(self) -> None:
         if self._logger:
             self._logger.info(f"SQLNotify Notifier started (sync mode, dialect: {self.dialect_name})")
 
+    @wrap_unhandled_error(lambda self: self._logger)
     def stop(self) -> None:
         """
         Stop watching for database changes synchronously
@@ -219,6 +222,7 @@ def stop(self) -> None:
 
         self._stop_sync()
 
+    @wrap_unhandled_error(lambda self: self._logger)
     async def astop(self) -> None:
         """
         Stop watching for database changes asynchronously.
@@ -261,6 +265,7 @@ def _stop_sync(self) -> None:
         if self._logger:
             self._logger.info("SQLNotify Notifier stopped (sync mode)")
 
+    @wrap_unhandled_error(lambda self: self._logger, reraise=False)
     def cleanup(self) -> None:
         """
         Remove all triggers and functions from the database synchronously
@@ -277,6 +282,7 @@ def cleanup(self) -> None:
 
         self._cleanup_sync()
 
+    @wrap_unhandled_error(lambda self: self._logger, reraise=False)
     async def acleanup(self) -> None:
         """
         Remove all triggers and functions from the database asynchronously.
@@ -311,6 +317,7 @@ def _cleanup_sync(self) -> None:
         if self._logger:
             self._logger.info("SQLNotify Notifier cleaned up all triggers and functions (sync mode)")
 
+    @wrap_unhandled_error(lambda self: self._logger)
     def notify(
         self,
         model: type | str,
@@ -345,6 +352,7 @@ def notify(
 
         self._notify_sync(watcher, payload, use_overflow_table)
 
+    @wrap_unhandled_error(lambda self: self._logger)
     async def anotify(
         self,
         model: type | str,
@@ -580,7 +588,7 @@ async def _dispatch_event(self, watcher: Watcher, payload: dict[str, Any]) -> No
 
         Args:
             watcher (Watcher): The watcher that triggered
-            payload (dict): The notification payload
+            payload (dict[str, Any]): The notification payload
         """
 
         if len(watcher.primary_keys) == 1:

src/sqlnotify/utils.py

@@ -1,21 +1,27 @@
+import inspect
+import logging
+from collections.abc import Callable
+from functools import wraps
+from typing import Any
+
 from sqlalchemy.engine import Engine
 from sqlalchemy.ext.asyncio import AsyncEngine
 
 from .constants import MAX_SQLNOTIFY_IDENTIFER_BYTES, MAX_SQLNOTIFY_PAYLOAD_BYTES
-from .exceptions import SQLNotifyIdentifierSizeError, SQLNotifyPayloadSizeError
+from .exceptions import SQLNotifyException, SQLNotifyIdentifierSizeError, SQLNotifyPayloadSizeError
 
 
 def extract_database_url(engine: AsyncEngine | Engine) -> str:
     """
     Extract the database URL from an engine for use with asyncpg LISTEN.
 
-    Converts SQLAlchemy URL format to plain PostgreSQL URL suitable for asyncpg.
+    Converts SQLAlchemy URL format to raw database DSN.
 
     Args:
         engine (Union[AsyncEngine, Engine]): SQLAlchemy Engine or AsyncEngine
 
     Returns:
-        str: PostgreSQL connection URL suitable for asyncpg
+        str: Database connection URL
     """
 
     url = engine.url.render_as_string(hide_password=False)
@@ -163,3 +169,65 @@ def validate_identifier_size(
         raise SQLNotifyIdentifierSizeError(combined)
 
     return True
+
+
+def wrap_unhandled_error(
+    logger_getter: Callable[..., logging.Logger] | None = None,
+    reraise=True,
+):
+    """
+    Decorator factory that wraps unhandled exceptions and converts them to SQLNotifyException
+    """
+
+    def _get_logger_from_call(*args, **kwargs):
+        if callable(logger_getter):
+            try:
+                return logger_getter(*args, **kwargs)
+            except Exception:
+                return None
+
+        return logger_getter
+
+    def decorator(func: Callable[..., Any]):
+
+        if inspect.iscoroutinefunction(func):
+
+            @wraps(func)
+            async def async_wrapper(*args, **kwargs):
+                try:
+                    return await func(*args, **kwargs)
+                except SQLNotifyException:
+                    raise
+                except Exception as e:
+                    logger = _get_logger_from_call(*args, **kwargs)
+                    if logger:
+                        logger.exception(f"Unhandled exception in {func.__name__}")
+
+                    if reraise:
+                        raise SQLNotifyException(f"SQLNotify unhandled exception caught. Error: {str(e)}") from e
+                    else:
+                        return None
+
+            return async_wrapper
+
+        else:
+
+            @wraps(func)
+            def sync_wrapper(*args, **kwargs):
+                try:
+                    return func(*args, **kwargs)
+                except SQLNotifyException:
+                    raise
+                except Exception as e:
+                    logger = _get_logger_from_call(*args, **kwargs)
+                    if logger:
+                        logger.exception(f"Unhandled exception in {func.__name__}")
+
+                    if reraise:
+                        raise SQLNotifyException(f"SQLNotify unhandled exception caught. Error: {str(e)}") from e
+                    else:
+                        return None
+
+            return sync_wrapper
+
+    return decorator