ADD: Live client stream exception warning

Author: nmacholl

CHANGELOG.md

@@ -4,9 +4,13 @@
 
 #### 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
 - Upgraded `databento-dbn` to 0.44.0
   - Added logic to set `code` when upgrading version 1 `SystemMsg` to newer versions
 
+#### Bug fixes
+- Streams opened by `Live.add_stream()` will now close properly when the streaming session is closed
+
 ## 0.65.0 - 2025-11-11
 
 #### Deprecations

databento/common/types.py

@@ -1,12 +1,19 @@
 import datetime as dt
+import logging
+import warnings
 from collections.abc import Callable
-from typing import Generic
+from os import PathLike
+import pathlib
+from typing import IO, Generic
 from typing import TypedDict
 from typing import TypeVar
 
 import databento_dbn
 import pandas as pd
 
+from databento.common.error import BentoWarning
+
+logger = logging.getLogger(__name__)
 
 DBNRecord = (
     databento_dbn.BBOMsg
@@ -88,3 +95,121 @@ class MappingIntervalDict(TypedDict):
     start_date: dt.date
     end_date: dt.date
     symbol: str
+
+
+class ClientStream:
+    def __init__(
+        self,
+        stream: IO[bytes] | PathLike[str] | str,
+        exc_fn: ExceptionCallback | None = None,
+        max_warnings: int = 10,
+    ) -> None:
+        is_managed = False
+
+        if isinstance(stream, (str, PathLike)):
+            stream = pathlib.Path(stream).open("xb")
+            is_managed = True
+
+        if not hasattr(stream, "write"):
+            raise ValueError(f"{type(stream).__name__} does not support write()")
+
+        if not hasattr(stream, "writable") or not stream.writable():
+            raise ValueError(f"{type(stream).__name__} is not a writable stream")
+
+        if exc_fn is not None and not callable(exc_fn):
+            raise ValueError(f"{exc_fn} is not callable")
+
+        self._stream = stream
+        self._exc_fn = exc_fn
+        self._max_warnings = max(0, max_warnings)
+        self._warning_count = 0
+        self._is_managed = is_managed
+
+    @property
+    def stream_name(self) -> str:
+        return getattr(self._stream, "__name__", str(self._stream))
+
+    @property
+    def is_closed(self) -> bool:
+        """
+        Return `True` if the underlying stream is closed.
+
+        Returns
+        -------
+        bool
+
+        """
+        return self._stream.closed
+
+    @property
+    def is_managed(self) -> bool:
+        """
+        Return `True` if the underlying stream was opened by the
+        `ClientStream`. This can be used to determine if the stream should be
+        closed automatically.
+
+        Returns
+        -------
+        bool
+
+        """
+        return self._is_managed
+
+    @property
+    def exc_callback_name(self) -> str:
+        return getattr(self._exc_fn, "__name__", str(self._exc_fn))
+
+    def close(self) -> None:
+        """
+        Close the underlying stream.
+        """
+        self._stream.close()
+
+    def flush(self) -> None:
+        """
+        Flush the underlying stream.
+        """
+        self._stream.flush()
+
+    def write(self, data: bytes) -> None:
+        """
+        Write data to the underlying stream. Any exceptions encountered will be
+        dispatched to the exception callback, if defined.
+
+        Parameters
+        ----------
+        data : bytes
+
+        """
+        try:
+            self._stream.write(data)
+        except Exception as exc:
+            if self._exc_fn is None:
+                self._warn(
+                    f"stream '{self.stream_name}' encountered an exception without an exception handler: {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.stream_name}'",
+                    BentoWarning,
+                    stacklevel=3,
+                )

databento/live/client.py

@@ -3,7 +3,6 @@
 import asyncio
 import logging
 import os
-import pathlib
 import queue
 import threading
 from collections.abc import Iterable
@@ -24,7 +23,7 @@
 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 DBNRecord
+from databento.common.types import ClientStream, DBNRecord
 from databento.common.types import ExceptionCallback
 from databento.common.types import ReconnectCallback
 from databento.common.types import RecordCallback
@@ -307,7 +306,9 @@ def add_stream(
             The IO stream to write to when handling live records as they arrive.
         exception_callback : Callable[[Exception], None], optional
             An error handling callback to process exceptions that are raised
-            when writing to the stream.
+            when writing to the stream. If no exception callback is provided,
+            any exceptions encountered will be logged and raised as warnings
+            for visibility.
 
         Raises
         ------
@@ -322,23 +323,12 @@ def add_stream(
         Live.add_callback
 
         """
-        if isinstance(stream, (str, PathLike)):
-            stream = pathlib.Path(stream).open("xb")
+        client_stream = ClientStream(stream=stream, exc_fn=exception_callback)
 
-        if not hasattr(stream, "write"):
-            raise ValueError(f"{type(stream).__name__} does not support write()")
-
-        if not hasattr(stream, "writable") or not stream.writable():
-            raise ValueError(f"{type(stream).__name__} is not a writable stream")
-
-        if exception_callback is not None and not callable(exception_callback):
-            raise ValueError(f"{exception_callback} is not callable")
-
-        stream_name = getattr(stream, "name", str(stream))
-        logger.info("adding user stream %s", stream_name)
+        logger.info("adding user stream %s", client_stream.stream_name)
         if self.metadata is not None:
-            stream.write(bytes(self.metadata))
-        self._session._user_streams.append((stream, exception_callback))
+            client_stream.write(self.metadata.encode())
+        self._session._user_streams.append(client_stream)
 
     def add_reconnect_callback(
         self,

databento/live/session.py

@@ -8,7 +8,6 @@
 import threading
 from collections.abc import Iterable
 from functools import partial
-from typing import IO
 from typing import Final
 
 import databento_dbn
@@ -20,7 +19,7 @@
 from databento.common.enums import ReconnectPolicy
 from databento.common.error import BentoError
 from databento.common.publishers import Dataset
-from databento.common.types import DBNRecord
+from databento.common.types import ClientStream, DBNRecord
 from databento.common.types import ExceptionCallback
 from databento.common.types import ReconnectCallback
 from databento.common.types import RecordCallback
@@ -148,6 +147,12 @@ class SessionMetadata:
     def __bool__(self) -> bool:
         return self.data is not None
 
+    @property
+    def has_ts_out(self) -> bool:
+        if self.data is None:
+            return False
+        return self.data.ts_out
+
     def check(self, other: databento_dbn.Metadata) -> None:
         """
         Verify the Metadata is compatible with another Metadata message. This
@@ -191,7 +196,7 @@ def __init__(
         dataset: Dataset | str,
         dbn_queue: DBNQueue,
         user_callbacks: list[tuple[RecordCallback, ExceptionCallback | None]],
-        user_streams: list[tuple[IO[bytes], ExceptionCallback | None]],
+        user_streams: list[ClientStream],
         loop: asyncio.AbstractEventLoop,
         metadata: SessionMetadata,
         ts_out: bool = False,
@@ -210,21 +215,15 @@ def received_metadata(self, metadata: databento_dbn.Metadata) -> None:
         if self._metadata:
             self._metadata.check(metadata)
         else:
-            metadata_bytes = metadata.encode()
-            for stream, exc_callback in self._user_streams:
+            for stream in self._user_streams:
                 try:
-                    stream.write(metadata_bytes)
+                    stream.write(metadata.encode())
                 except Exception as exc:
-                    stream_name = getattr(stream, "name", str(stream))
                     logger.error(
-                        "error writing %d bytes to `%s` stream",
-                        len(metadata_bytes),
-                        stream_name,
+                        "error writing metadata to `%s` stream",
+                        stream.stream_name,
                         exc_info=exc,
                     )
-                    if exc_callback is not None:
-                        exc_callback(exc)
-
             self._metadata.data = metadata
         return super().received_metadata(metadata)
 
@@ -252,26 +251,20 @@ def _dispatch_callbacks(self, record: DBNRecord) -> None:
                     exc_callback(exc)
 
     def _dispatch_writes(self, record: DBNRecord) -> None:
-        if hasattr(record, "ts_out"):
-            ts_out_bytes = struct.pack("Q", record.ts_out)
-        else:
-            ts_out_bytes = b""
-
-        record_bytes = bytes(record) + ts_out_bytes
-
-        for stream, exc_callback in self._user_streams:
+        record_bytes = bytes(record)
+        ts_out_bytes = struct.pack("Q", record.ts_out) if self._metadata.has_ts_out else b""
+        for stream in self._user_streams:
             try:
                 stream.write(record_bytes)
+                stream.write(ts_out_bytes)
             except Exception as exc:
-                stream_name = getattr(stream, "name", str(stream))
                 logger.error(
-                    "error writing %d bytes to `%s` stream",
-                    len(record_bytes),
-                    stream_name,
+                    "error writing %s record (%d bytes) to `%s` stream",
+                    type(record).__name__,
+                    len(record_bytes) + len(ts_out_bytes),
+                    stream.stream_name,
                     exc_info=exc,
                 )
-                if exc_callback is not None:
-                    exc_callback(exc)
 
     def _queue_for_iteration(self, record: DBNRecord) -> None:
         self._dbn_queue.put(record)
@@ -323,7 +316,7 @@ def __init__(
         self._metadata = SessionMetadata()
         self._user_gateway: str | None = user_gateway
         self._user_callbacks: list[tuple[RecordCallback, ExceptionCallback | None]] = []
-        self._user_streams: list[tuple[IO[bytes], ExceptionCallback | None]] = []
+        self._user_streams: list[ClientStream] = []
         self._user_reconnect_callbacks: list[tuple[ReconnectCallback, ExceptionCallback | None]] = (
             []
         )
@@ -551,10 +544,11 @@ async def wait_for_close(self) -> None:
     def _cleanup(self) -> None:
         logger.debug("cleaning up session_id=%s", self.session_id)
         self._user_callbacks.clear()
-        for item in self._user_streams:
-            stream, _ = item
-            if not stream.closed:
+        for stream in self._user_streams:
+            if not stream.is_closed:
                 stream.flush()
+            if stream.is_managed:
+                stream.close()
 
         self._user_callbacks.clear()
         self._user_streams.clear()

tests/test_live_client.py

@@ -1008,7 +1008,7 @@ def test_live_add_stream(
 
     # Assert
     assert len(live_client._session._user_streams) == 1
-    assert (stream, None) in live_client._session._user_streams
+    assert stream == live_client._session._user_streams[0]._stream
 
 
 def test_live_add_stream_invalid(