FIX: Fix writing on timeseries.get_range error

Author: nmacholl

CHANGELOG.md

@@ -2,11 +2,12 @@
 
 ## 0.14.0 - TBD
 - Added support for reusing a `Live` client to reconnect
-- Changed iteration of `Live` to no longer yield DBN metadata
-- Changed `Live` callbacks to no longer yield DBN metadata
 - Added `metadata` property to `Live`
 - Added `DatatbentoLiveProtocol` class
 - Added support for emitting warnings in API response headers
+- Changed iteration of `Live` to no longer yield DBN metadata
+- Changed `Live` callbacks to no longer yield DBN metadata
+- Fixed issue where `Historical.timeseries.get_range` would write empty files on error
 - Fixed issue with `numpy` types not being handled in symbols field
 - Upgraded `aiohttp` to 3.8.3
 - Upgraded `numpy` to to 1.23.5

databento/__init__.py

@@ -1,48 +1,43 @@
 import logging
 import warnings
 
+from databento_dbn import ErrorMsg
+from databento_dbn import ImbalanceMsg
+from databento_dbn import InstrumentDefMsg
+from databento_dbn import MBOMsg
+from databento_dbn import MBP1Msg
+from databento_dbn import MBP10Msg
+from databento_dbn import Metadata
+from databento_dbn import OHLCVMsg
+from databento_dbn import StatMsg
+from databento_dbn import SymbolMappingMsg
+from databento_dbn import SystemMsg
+from databento_dbn import TradeMsg
+
 from databento.common import bentologging
 from databento.common.dbnstore import DBNStore
-from databento.common.enums import (
-    Compression,
-    Dataset,
-    Delivery,
-    Encoding,
-    FeedMode,
-    HistoricalGateway,
-    Packaging,
-    RecordFlags,
-    RollRule,
-    Schema,
-    SplitDuration,
-    SType,
-    SymbologyResolution,
-)
-from databento.common.error import (
-    BentoClientError,
-    BentoError,
-    BentoHttpError,
-    BentoServerError,
-)
+from databento.common.enums import Compression
+from databento.common.enums import Dataset
+from databento.common.enums import Delivery
+from databento.common.enums import Encoding
+from databento.common.enums import FeedMode
+from databento.common.enums import HistoricalGateway
+from databento.common.enums import Packaging
+from databento.common.enums import RecordFlags
+from databento.common.enums import RollRule
+from databento.common.enums import Schema
+from databento.common.enums import SplitDuration
+from databento.common.enums import SType
+from databento.common.enums import SymbologyResolution
+from databento.common.error import BentoClientError
+from databento.common.error import BentoError
+from databento.common.error import BentoHttpError
+from databento.common.error import BentoServerError
 from databento.historical.api import API_VERSION
 from databento.historical.client import Historical
 from databento.live import DBNRecord
 from databento.live.client import Live
 from databento.version import __version__  # noqa
-from databento_dbn import (
-    ErrorMsg,
-    ImbalanceMsg,
-    InstrumentDefMsg,
-    MBOMsg,
-    MBP1Msg,
-    MBP10Msg,
-    Metadata,
-    OHLCVMsg,
-    StatMsg,
-    SymbolMappingMsg,
-    SystemMsg,
-    TradeMsg,
-)
 
 
 __all__ = [

databento/common/data.py

@@ -1,6 +1,7 @@
 from typing import Dict, List, Tuple, Union
 
 import numpy as np
+
 from databento.common.enums import Schema
 
 

databento/common/dbnstore.py

@@ -495,7 +495,8 @@ def _map_symbols(self, df: pd.DataFrame, pretty_ts: bool) -> pd.DataFrame:
             df_index = df.index if pretty_ts else pd.to_datetime(df.index, utc=True)
             dates = [ts.date() for ts in df_index]
             df["symbol"] = [
-                self._instrument_id_index[dates[i]][p] for i, p in enumerate(df["instrument_id"])
+                self._instrument_id_index[dates[i]][p]
+                for i, p in enumerate(df["instrument_id"])
             ]
 
         return df

databento/common/validation.py

@@ -4,7 +4,8 @@
 from os import PathLike
 from pathlib import Path
 from typing import Optional, Type, TypeVar, Union
-from urllib.parse import urlsplit, urlunsplit
+from urllib.parse import urlsplit
+from urllib.parse import urlunsplit
 
 
 E = TypeVar("E", bound=Enum)

databento/historical/api/timeseries.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 from datetime import date
-from io import BufferedIOBase
-from io import BytesIO
 from os import PathLike
 from typing import List, Optional, Tuple, Union
 
@@ -118,24 +116,13 @@ def get_range(
         if limit is not None:
             params.append(("limit", str(limit)))
 
-        if path is not None:
-            writer: BufferedIOBase = open(path, "x+b")
-        else:
-            writer = BytesIO()
-
-        self._stream(
+        return self._stream(
             url=self._base_url + ".get_range",
             params=params,
             basic_auth=True,
-            writer=writer,
+            path=path,
         )
 
-        if path is not None:
-            writer.close()
-            return DBNStore.from_file(path)
-        writer.seek(0)  # rewind for read
-        return DBNStore.from_bytes(writer.read())
-
     async def get_range_async(
         self,
         dataset: Union[Dataset, str],
@@ -218,23 +205,13 @@ async def get_range_async(
             ("compression", str(Compression.ZSTD)),  # Always request zstd
         ]
 
+        # Optional Parameters
         if limit is not None:
             params.append(("limit", str(limit)))
 
-        if path is not None:
-            writer: BufferedIOBase = open(path, "x+b")
-        else:
-            writer = BytesIO()
-
-        await self._stream_async(
+        return await self._stream_async(
             url=self._base_url + ".get_range",
             params=params,
             basic_auth=True,
-            writer=writer,
+            path=path,
         )
-
-        if path is not None:
-            writer.close()
-            return DBNStore.from_file(path)
-        writer.seek(0)  # rewind for read
-        return DBNStore.from_bytes(writer.read())

databento/historical/http.py

@@ -1,9 +1,12 @@
+from __future__ import annotations
+
 import json
 import sys
 import warnings
-from io import BufferedIOBase
+from io import BytesIO
 from json.decoder import JSONDecodeError
-from typing import Any, List, Optional, Tuple, Union
+from os import PathLike
+from typing import IO, Any, List, Optional, Tuple, Union
 
 import aiohttp
 import requests
@@ -12,6 +15,7 @@
 from requests import Response
 from requests.auth import HTTPBasicAuth
 
+from databento.common.dbnstore import DBNStore
 from databento.common.error import BentoClientError
 from databento.common.error import BentoDeprecationWarning
 from databento.common.error import BentoServerError
@@ -108,8 +112,8 @@ def _stream(
         url: str,
         params: List[Tuple[str, Optional[str]]],
         basic_auth: bool,
-        writer: BufferedIOBase,
-    ) -> None:
+        path: Optional[Union[PathLike[str], str]] = None,
+    ) -> DBNStore:
         self._check_api_key()
 
         with requests.get(
@@ -123,16 +127,28 @@ def _stream(
             check_backend_warnings(response)
             check_http_error(response)
 
+            if path is None:
+                writer: IO[bytes] = BytesIO()
+            else:
+                writer = open(path, "x+b")
+
             for chunk in response.iter_content(chunk_size=_32KB):
                 writer.write(chunk)
 
+            if path is None:
+                writer.seek(0)
+                return DBNStore.from_bytes(writer)
+
+            writer.close()
+            return DBNStore.from_file(path)
+
     async def _stream_async(
         self,
         url: str,
         params: List[Tuple[str, Optional[str]]],
         basic_auth: bool,
-        writer: BufferedIOBase,
-    ) -> None:
+        path: Optional[Union[PathLike[str], str]] = None,
+    ) -> DBNStore:
         self._check_api_key()
 
         async with aiohttp.ClientSession() as session:
@@ -148,9 +164,21 @@ async def _stream_async(
                 check_backend_warnings(response)
                 await check_http_error_async(response)
 
+                if path is None:
+                    writer: IO[bytes] = BytesIO()
+                else:
+                    writer = open(path, "x+b")
+
                 async for chunk in response.content.iter_chunks():
                     writer.write(chunk[0])
 
+                if path is None:
+                    writer.seek(0)
+                    return DBNStore.from_bytes(writer)
+
+                writer.close()
+                return DBNStore.from_file(path)
+
 
 def is_400_series_error(status: int) -> bool:
     return status // 100 == 4

databento/live/protocol.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 import asyncio
 import logging
-from functools import singledispatch
-from functools import update_wrapper
+from collections.abc import Iterable
+from functools import singledispatchmethod
 from numbers import Number
-from typing import Any, Callable, Iterable, Optional, Union
+from typing import Optional, Union
 
 import databento_dbn
 
@@ -46,32 +48,6 @@
 logger = logging.getLogger(__name__)
 
 
-def singledispatchmethod(
-    func: Callable[..., Any],
-) -> Any:
-    """
-    Decorate a function to dispatch arguments by type.
-    This is a custom implementation of functools.singledispatchmethod.
-
-    See Also
-    --------
-    functools.singledispatch
-
-    Notes
-    -----
-    This should be removed when python 3.7 is no longer supported.
-
-    """
-    dispatcher: Any = singledispatch(func)
-
-    def _wrapper(*args: object, **kw: object) -> Any:
-        return dispatcher.dispatch(args[1].__class__)(*args, **kw)
-
-    setattr(_wrapper, "register", getattr(dispatcher, "register"))
-    update_wrapper(_wrapper, func)
-    return _wrapper
-
-
 class DatabentoLiveProtocol(asyncio.BufferedProtocol):
     """
     A BufferedProtocol implementation for the Databento live subscription

examples/historical_symbology_resolve.py

@@ -1,6 +1,7 @@
 from pprint import pprint
 
-from databento import Historical, SType
+from databento import Historical
+from databento import SType
 
 
 if __name__ == "__main__":

examples/historical_timeseries_async.py

@@ -1,7 +1,8 @@
 import asyncio
 from pprint import pprint
 
-from databento import DBNStore, Historical
+from databento import DBNStore
+from databento import Historical
 
 
 async def example_get_range_async() -> None: