ADD: Live client callback exception warning

Author: nmacholl

CHANGELOG.md

@@ -5,6 +5,7 @@
 #### Enhancements
 - Added a property `Live.session_id` which returns the streaming session ID when the client is connected
 - Streams added with `Live.add_stream()` which do not define an exception handler will now emit a warning if an exception is raised while executing the callback
+- Callback functions added with `Live.add_callback()` which do not define an exception handler will now emit a warning if an exception is raised while executing the callback
 - Upgraded `databento-dbn` to 0.44.0
   - Added logic to set `code` when upgrading version 1 `SystemMsg` to newer versions
 

databento/common/types.py

@@ -1,12 +1,13 @@
 import datetime as dt
 import logging
-import warnings
 from collections.abc import Callable
 from os import PathLike
 import pathlib
-from typing import IO, Generic
+from typing import Generic
+from typing import IO
 from typing import TypedDict
 from typing import TypeVar
+import warnings
 
 import databento_dbn
 import pandas as pd
@@ -39,10 +40,6 @@
     | databento_dbn.ErrorMsgV1
 )
 
-RecordCallback = Callable[[DBNRecord], None]
-ExceptionCallback = Callable[[Exception], None]
-ReconnectCallback = Callable[[pd.Timestamp, pd.Timestamp], None]
-
 _T = TypeVar("_T")
 
 
@@ -97,6 +94,11 @@ class MappingIntervalDict(TypedDict):
     symbol: str
 
 
+RecordCallback = Callable[[DBNRecord], None]
+ExceptionCallback = Callable[[Exception], None]
+ReconnectCallback = Callable[[pd.Timestamp, pd.Timestamp], None]
+
+
 class ClientStream:
     def __init__(
         self,
@@ -213,3 +215,73 @@ def _warn(self, msg: str) -> None:
                     BentoWarning,
                     stacklevel=3,
                 )
+
+
+class ClientRecordCallback:
+    def __init__(
+        self,
+        fn: RecordCallback,
+        exc_fn: ExceptionCallback | None = None,
+        max_warnings: int = 10,
+    ) -> None:
+        if not callable(fn):
+            raise ValueError(f"{fn} is not callable")
+        if exc_fn is not None and not callable(exc_fn):
+            raise ValueError(f"{exc_fn} is not callable")
+
+        self._fn = fn
+        self._exc_fn = exc_fn
+        self._max_warnings = max(0, max_warnings)
+        self._warning_count = 0
+
+    @property
+    def callback_name(self) -> str:
+        return getattr(self._fn, "__name__", str(self._fn))
+
+    @property
+    def exc_callback_name(self) -> str:
+        return getattr(self._exc_fn, "__name__", str(self._exc_fn))
+
+    def call(self, record: DBNRecord) -> None:
+        """
+        Execute the callback function, passing `record` in as the first
+        argument. Any exceptions encountered will be dispatched to the
+        exception callback, if defined.
+
+        Parameters
+        ----------
+        record : DBNRecord
+
+        """
+        try:
+            self._fn(record)
+        except Exception as exc:
+            if self._exc_fn is None:
+                self._warn(
+                    f"callback '{self.callback_name}' encountered an exception without an exception callback: {repr(exc)}",
+                )
+            else:
+                try:
+                    self._exc_fn(exc)
+                except Exception as inner_exc:
+                    self._warn(
+                        f"exception callback '{self.exc_callback_name}' encountered an exception: {repr(inner_exc)}",
+                    )
+                    raise inner_exc from exc
+            raise exc
+
+    def _warn(self, msg: str) -> None:
+        logger.warning(msg)
+        if self._warning_count < self._max_warnings:
+            self._warning_count += 1
+            warnings.warn(
+                msg,
+                BentoWarning,
+                stacklevel=3,
+            )
+            if self._warning_count == self._max_warnings:
+                warnings.warn(
+                    f"suppressing further warnings for '{self.callback_name}'",
+                    BentoWarning,
+                    stacklevel=3,
+                )

databento/live/client.py

@@ -23,7 +23,9 @@
 from databento.common.error import BentoError
 from databento.common.parsing import optional_datetime_to_unix_nanoseconds
 from databento.common.publishers import Dataset
-from databento.common.types import ClientStream, DBNRecord
+from databento.common.types import ClientRecordCallback
+from databento.common.types import ClientStream
+from databento.common.types import DBNRecord
 from databento.common.types import ExceptionCallback
 from databento.common.types import ReconnectCallback
 from databento.common.types import RecordCallback
@@ -110,7 +112,7 @@ def __init__(
             reconnect_policy=reconnect_policy,
         )
 
-        self._session._user_callbacks.append((self._map_symbol, None))
+        self._session._user_callbacks.append(ClientRecordCallback(self._map_symbol))
 
         with Live._lock:
             if not Live._thread.is_alive():
@@ -269,7 +271,9 @@ def add_callback(
             A callback to register for handling live records as they arrive.
         exception_callback : Callable[[Exception], None], optional
             An error handling callback to process exceptions that are raised
-            in `record_callback`.
+            in `record_callback`. If no exception callback is provided,
+            any exceptions encountered will be logged and raised as warnings
+            for visibility.
 
         Raises
         ------
@@ -282,15 +286,13 @@ def add_callback(
         Live.add_stream
 
         """
-        if not callable(record_callback):
-            raise ValueError(f"{record_callback} is not callable")
-
-        if exception_callback is not None and not callable(exception_callback):
-            raise ValueError(f"{exception_callback} is not callable")
+        client_callback = ClientRecordCallback(
+            fn=record_callback,
+            exc_fn=exception_callback,
+        )
 
-        callback_name = getattr(record_callback, "__name__", str(record_callback))
-        logger.info("adding user callback %s", callback_name)
-        self._session._user_callbacks.append((record_callback, exception_callback))
+        logger.info("adding user callback %s", client_callback.callback_name)
+        self._session._user_callbacks.append(client_callback)
 
     def add_stream(
         self,

databento/live/session.py

@@ -19,10 +19,11 @@
 from databento.common.enums import ReconnectPolicy
 from databento.common.error import BentoError
 from databento.common.publishers import Dataset
-from databento.common.types import ClientStream, DBNRecord
+from databento.common.types import ClientRecordCallback
+from databento.common.types import ClientStream
+from databento.common.types import DBNRecord
 from databento.common.types import ExceptionCallback
 from databento.common.types import ReconnectCallback
-from databento.common.types import RecordCallback
 from databento.live.gateway import SubscriptionRequest
 from databento.live.protocol import DatabentoLiveProtocol
 
@@ -195,8 +196,8 @@ def __init__(
         api_key: str,
         dataset: Dataset | str,
         dbn_queue: DBNQueue,
-        user_callbacks: list[tuple[RecordCallback, ExceptionCallback | None]],
         user_streams: list[ClientStream],
+        user_callbacks: list[ClientRecordCallback],
         loop: asyncio.AbstractEventLoop,
         metadata: SessionMetadata,
         ts_out: bool = False,
@@ -237,18 +238,16 @@ def received_record(self, record: DBNRecord) -> None:
         return super().received_record(record)
 
     def _dispatch_callbacks(self, record: DBNRecord) -> None:
-        for callback, exc_callback in self._user_callbacks:
+        for callback in self._user_callbacks:
             try:
-                callback(record)
+                callback.call(record)
             except Exception as exc:
                 logger.error(
                     "error dispatching %s to `%s` callback",
                     type(record).__name__,
-                    getattr(callback, "__name__", str(callback)),
+                    callback.callback_name,
                     exc_info=exc,
                 )
-                if exc_callback is not None:
-                    exc_callback(exc)
 
     def _dispatch_writes(self, record: DBNRecord) -> None:
         record_bytes = bytes(record)
@@ -315,8 +314,8 @@ def __init__(
         self._loop = loop
         self._metadata = SessionMetadata()
         self._user_gateway: str | None = user_gateway
-        self._user_callbacks: list[tuple[RecordCallback, ExceptionCallback | None]] = []
         self._user_streams: list[ClientStream] = []
+        self._user_callbacks: list[ClientRecordCallback] = []
         self._user_reconnect_callbacks: list[tuple[ReconnectCallback, ExceptionCallback | None]] = (
             []
         )
@@ -527,19 +526,20 @@ async def wait_for_close(self) -> None:
             return
 
         try:
-            await self._protocol.authenticated
-        except Exception as exc:
-            raise BentoError(exc) from None
-
-        try:
-            if self._reconnect_task is not None:
-                await self._reconnect_task
-            else:
-                await self._protocol.disconnected
-        except Exception as exc:
-            raise BentoError(exc) from None
+            try:
+                await self._protocol.authenticated
+            except Exception as exc:
+                raise BentoError(exc) from None
 
-        self._cleanup()
+            try:
+                if self._reconnect_task is not None:
+                    await self._reconnect_task
+                else:
+                    await self._protocol.disconnected
+            except Exception as exc:
+                raise BentoError(exc) from None
+        finally:
+            self._cleanup()
 
     def _cleanup(self) -> None:
         logger.debug("cleaning up session_id=%s", self.session_id)

tests/test_live_client.py

@@ -983,15 +983,15 @@ def test_live_add_callback(
     """
 
     # Arrange
-    def callback(_: object) -> None:
+    def test_callback(_: object) -> None:
         pass
 
     # Act
-    live_client.add_callback(callback)
+    live_client.add_callback(test_callback)
 
     # Assert
     assert len(live_client._session._user_callbacks) == 2  # include map_symbols callback
-    assert (callback, None) in live_client._session._user_callbacks
+    assert live_client._session._user_callbacks[-1].callback_name == "test_callback"
 
 
 def test_live_add_stream(