VER: Release 0.36.1

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,19 @@
 # Changelog
 
+## 0.36.1 - 2025-06-17
+
+### Bug fixes
+- Updated conversion error messages in `TryFrom<RecordRef> for RecordRefEnum` for
+  release of DBN version 3
+- Updated `TryFrom<RecordRef> for RecordRefEnum` for versioning of `StatMsg`,
+  `SystemMsg`, and  `ErrorMsg`
+- Fixed conversion of `BboMsg` in `TryFrom<RecordRef> for RecordRefEnum`
+- Implemented `AsyncDecodeRecordRef` and `AsyncDecodeRecord` for `AsyncDecoder` and
+  `AsyncRecordDecoder`
+- Fixed setting of `ts_out` property of `DbnFsm` based on decoded metadata. This
+  was preventing `ts_out` from being correctly decoded in the Python `DBNDecoder`
+- Fixed decoding of `ts_out` with first records in `DBNDecoder`
+
 ## 0.36.0 - 2025-06-10
 
 ### Enhancements
@@ -191,8 +205,8 @@ upgrading the input to DBNv3.
 ## 0.29.0 - 2025-03-17
 
 ### Enhancements
-- Added new venues, datasets, and publishers for ICE Futures US, ICE Futures Europe
-  (Financial products), Eurex, and European Energy Exchange (EEX)
+- Added new venues, datasets, and publishers for ICE Futures US, ICE Europe Financials
+  products, Eurex, and European Energy Exchange (EEX)
 - Added new `SkipBytes` and `AsyncSkipBytes` traits which are a subset of the `Seek`
   and `AsyncSeek` traits respectively, only supporting seeking forward from the current
   position

Cargo.lock

@@ -11,7 +11,7 @@ resolver = "2"
 [workspace.package]
 authors = ["Databento <[email protected]>"]
 edition = "2021"
-version = "0.36.0"
+version = "0.36.1"
 documentation = "https://databento.com/docs"
 repository = "https://github.com/databento/dbn"
 license = "Apache-2.0"

Cargo.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "databento-dbn"
-version = "0.36.0"
+version = "0.36.1"
 description = "Python bindings for encoding and decoding Databento Binary Encoding (DBN)"
 authors = ["Databento <[email protected]>"]
 license = "Apache-2.0"
@@ -17,7 +17,7 @@ build-backend = "maturin"
 
 [project]
 name = "databento-dbn"
-version = "0.36.0"
+version = "0.36.1"
 authors = [
     { name = "Databento", email = "[email protected]" }
 ]

python/pyproject.toml

@@ -47,13 +47,14 @@ impl DbnDecoder {
     }
 
     fn decode(&mut self) -> PyResult<Vec<PyObject>> {
-        let ts_out = self.fsm.ts_out();
+        let mut ts_out = self.fsm.ts_out();
         let mut py_recs = Vec::new();
         loop {
             let mut rec_refs = Vec::new();
             match self.fsm.process_all(&mut rec_refs, None) {
                 ProcessResult::ReadMore(_) => return Ok(py_recs),
                 ProcessResult::Metadata(metadata) => {
+                    ts_out = self.fsm.ts_out();
                     py_recs.push(Python::with_gil(|py| metadata.into_py_any(py))?)
                 }
                 ProcessResult::Record(_) => {
@@ -254,6 +255,44 @@ except Exception:
         }).unwrap();
     }
 
+    #[test]
+    fn test_dbn_decoder_ts_out() {
+        setup();
+        Python::with_gil(|py| {
+            Python::run(
+                py,
+                c_str!(
+                    r#"from _lib import DBNDecoder, DBNError, Metadata, Schema, SType
+
+metadata = Metadata(
+    version=3,
+    dataset="IFUS.IMPACT",
+    schema=Schema.MBO,
+    start=1,
+    stype_in=SType.RAW_SYMBOL,
+    stype_out=SType.INSTRUMENT_ID,
+    end=2,
+    ts_out=True,
+    symbols=[],
+    partial=[],
+    not_found=[],
+    mappings=[]
+)
+metadata_bytes = bytes(metadata)
+decoder = DBNDecoder()
+decoder.write(metadata_bytes)
+records = decoder.decode()
+for record in records:
+    assert record.ts_out is not None
+"#
+                ),
+                None,
+                None,
+            )
+        })
+        .unwrap();
+    }
+
     #[test]
     fn test_dbn_decoder_no_metadata() {
         setup();

python/src/dbn_decoder.rs

@@ -16,7 +16,7 @@ name = "dbn"
 path = "src/main.rs"
 
 [dependencies]
-dbn = { path = "../dbn", version = "=0.36.0", default-features = false }
+dbn = { path = "../dbn", version = "=0.36.1", default-features = false }
 
 anyhow = { workspace = true }
 clap = { version = "4.5", features = ["derive", "wrap_help"] }

rust/dbn-cli/Cargo.toml

@@ -25,7 +25,7 @@ serde = ["dep:serde", "time/parsing", "time/serde"]
 trivial_copy = []
 
 [dependencies]
-dbn-macros = { version = "=0.36.0", path = "../dbn-macros" }
+dbn-macros = { version = "=0.36.1", path = "../dbn-macros" }
 
 async-compression = { version = "0.4.23", features = ["tokio", "zstd"], optional = true }
 csv = { workspace = true }

rust/dbn/Cargo.toml

@@ -30,7 +30,7 @@ pub use stream::StreamIterDecoder;
 
 use std::{io::Seek, mem};
 
-use crate::{HasRType, Metadata, Record, RecordRef, VersionUpgradePolicy};
+use crate::{HasRType, Metadata, RecordRef, VersionUpgradePolicy};
 
 /// Trait for types that decode references to DBN records of a dynamic type.
 pub trait DecodeRecordRef {
@@ -94,22 +94,6 @@ pub trait DecodeRecord {
     }
 }
 
-fn decode_record_from_ref<T: HasRType>(rec_ref: Option<RecordRef>) -> crate::Result<Option<&T>> {
-    if let Some(rec_ref) = rec_ref {
-        rec_ref
-            .get::<T>()
-            .ok_or_else(|| {
-                crate::Error::conversion::<T>(format!(
-                    "record with rtype {:#04X}",
-                    rec_ref.header().rtype
-                ))
-            })
-            .map(Some)
-    } else {
-        Ok(None)
-    }
-}
-
 /// A trait alias for DBN decoders with metadata.
 pub trait DecodeDbn: DecodeRecord + DecodeRecordRef + DbnMetadata {}
 
@@ -199,7 +183,7 @@ pub trait AsyncDecodeRecord {
     /// # Cancel safety
     /// This method is not cancellation safe. If used within a `tokio::select!` statement
     /// partially decoded records will be lost and the stream may be corrupted.
-    async fn decode_records<T: HasRType + Clone>(mut self) -> crate::Result<Vec<T>>
+    async fn decode_records<T: HasRType + Clone>(&mut self) -> crate::Result<Vec<T>>
     where
         Self: Sized,
     {
@@ -277,13 +261,13 @@ impl FromLittleEndianSlice for u16 {
     }
 }
 
-#[cfg(test)]
-mod tests {
-    pub const TEST_DATA_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../tests/data");
-}
-
 #[cfg(feature = "async")]
 pub use self::dbn::{
     AsyncDecoder as AsyncDbnDecoder, AsyncMetadataDecoder as AsyncDbnMetadataDecoder,
     AsyncRecordDecoder as AsyncDbnRecordDecoder,
 };
+
+#[cfg(test)]
+mod tests {
+    pub const TEST_DATA_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../tests/data");
+}

rust/dbn/src/decode.rs

@@ -10,9 +10,10 @@ use crate::{
     decode::{
         dbn::fsm::{DbnFsm, ProcessResult},
         zstd::zstd_decoder,
-        AsyncSkipBytes, DbnMetadata, VersionUpgradePolicy, ZSTD_FILE_BUFFER_CAPACITY,
+        AsyncDecodeRecord, AsyncDecodeRecordRef, AsyncSkipBytes, DbnMetadata, VersionUpgradePolicy,
+        ZSTD_FILE_BUFFER_CAPACITY,
     },
-    HasRType, Metadata, Record, RecordRef, Result, DBN_VERSION,
+    HasRType, Metadata, RecordRef, Result, DBN_VERSION,
 };
 
 /// An async decoder for Databento Binary Encoding (DBN), both metadata and records.
@@ -240,6 +241,24 @@ where
     }
 }
 
+impl<R> AsyncDecodeRecordRef for Decoder<R>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    async fn decode_record_ref(&mut self) -> crate::Result<Option<RecordRef>> {
+        self.decoder.decode_ref().await
+    }
+}
+
+impl<R> AsyncDecodeRecord for Decoder<R>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    async fn decode_record<'a, T: HasRType + 'a>(&'a mut self) -> crate::Result<Option<&'a T>> {
+        self.decoder.decode().await
+    }
+}
+
 /// An async decoder for files and streams of Databento Binary Encoding (DBN) records.
 pub struct RecordDecoder<R>
 where
@@ -343,20 +362,13 @@ where
     /// This method is cancel safe. It can be used within a `tokio::select!` statement
     /// without the potential for corrupting the input stream.
     pub async fn decode<'a, T: HasRType + 'a>(&'a mut self) -> Result<Option<&'a T>> {
-        let rec_ref = self.decode_ref().await?;
-        if let Some(rec_ref) = rec_ref {
-            rec_ref
-                .get::<T>()
-                .ok_or_else(|| {
-                    crate::Error::conversion::<T>(format!(
-                        "record with rtype {:#04X}",
-                        rec_ref.header().rtype
-                    ))
-                })
-                .map(Some)
-        } else {
-            Ok(None)
-        }
+        self.decode_ref().await.and_then(|rec| {
+            if let Some(rec) = rec {
+                rec.try_get().map(Some)
+            } else {
+                Ok(None)
+            }
+        })
     }
 
     /// Tries to decode all records into a `Vec`. This eagerly decodes the data.
@@ -428,6 +440,24 @@ where
     }
 }
 
+impl<R> AsyncDecodeRecordRef for RecordDecoder<R>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    async fn decode_record_ref(&mut self) -> crate::Result<Option<RecordRef>> {
+        self.decode_ref().await
+    }
+}
+
+impl<R> AsyncDecodeRecord for RecordDecoder<R>
+where
+    R: io::AsyncReadExt + Unpin,
+{
+    async fn decode_record<'a, T: HasRType + 'a>(&'a mut self) -> crate::Result<Option<&'a T>> {
+        self.decode().await
+    }
+}
+
 impl<R> RecordDecoder<R>
 where
     R: AsyncSkipBytes + io::AsyncReadExt + Unpin,
@@ -665,8 +695,8 @@ mod tests {
             AsyncEncodeRecord, DbnEncodable,
         },
         rtype, v1, v2, Bbo1SMsg, CbboMsg, Cmbp1Msg, Error, ErrorMsg, ImbalanceMsg,
-        InstrumentDefMsg, MboMsg, Mbp10Msg, Mbp1Msg, OhlcvMsg, RecordHeader, Result, Schema,
-        StatMsg, StatusMsg, TbboMsg, TradeMsg, WithTsOut,
+        InstrumentDefMsg, MboMsg, Mbp10Msg, Mbp1Msg, OhlcvMsg, Record, RecordHeader, Result,
+        Schema, StatMsg, StatusMsg, TbboMsg, TradeMsg, WithTsOut,
     };
 
     #[rstest]