VER: Release 0.33.2

Author: threecgreen

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 0.33.2 - 2025-05-06
+
+### Enhancements
+- Added `with_compression_level` methods to `DynWriter`, `AsyncDynWriter`, and
+  `AsyncDynBufWriter`
+
+### Bug fixes
+- Fixed `AsyncDynBufWriter` interface
+
 ## 0.33.1 - 2025-04-29
 
 ### Enhancements

Cargo.lock

@@ -11,7 +11,7 @@ resolver = "2"
 [workspace.package]
 authors = ["Databento <[email protected]>"]
 edition = "2021"
-version = "0.33.1"
+version = "0.33.2"
 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.33.1"
+version = "0.33.2"
 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.33.1"
+version = "0.33.2"
 authors = [
     { name = "Databento", email = "[email protected]" }
 ]

python/pyproject.toml

@@ -16,7 +16,7 @@ name = "dbn"
 path = "src/main.rs"
 
 [dependencies]
-dbn = { path = "../dbn", version = "=0.33.1", default-features = false }
+dbn = { path = "../dbn", version = "=0.33.2", 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.33.1", path = "../dbn-macros" }
+dbn-macros = { version = "=0.33.2", path = "../dbn-macros" }
 
 async-compression = { version = "0.4.23", features = ["tokio", "zstd"], optional = true }
 csv = { workspace = true }

rust/dbn/Cargo.toml

@@ -176,8 +176,7 @@ pub trait InstrumentDefRec: HasRType {
     /// This function returns an error if `asset` contains invalid UTF-8.
     fn asset(&self) -> crate::Result<&str>;
 
-    /// Returns the type of the strument, e.g. FUT for future or future spread as
-    /// a `&str`.
+    /// Returns the [Security type](https://databento.com/docs/schemas-and-data-formats/instrument-definitions#security-type) of the instrument, e.g. FUT for future or future spread as a `&str`.
     ///
     /// # Errors
     /// This function returns an error if `security_type` contains invalid UTF-8.

rust/dbn/src/compat.rs

@@ -202,8 +202,15 @@ pub trait EncodeRecordTextExt: EncodeRecord + EncodeRecordRef {
 pub const ZSTD_COMPRESSION_LEVEL: i32 = 0;
 
 fn zstd_encoder<'a, W: io::Write>(writer: W) -> Result<zstd::stream::AutoFinishEncoder<'a, W>> {
-    let mut zstd_encoder = zstd::Encoder::new(writer, ZSTD_COMPRESSION_LEVEL)
-        .map_err(|e| Error::io(e, "creating zstd encoder"))?;
+    zstd_encoder_with_clevel(writer, ZSTD_COMPRESSION_LEVEL)
+}
+
+fn zstd_encoder_with_clevel<'a, W: io::Write>(
+    writer: W,
+    level: i32,
+) -> Result<zstd::stream::AutoFinishEncoder<'a, W>> {
+    let mut zstd_encoder =
+        zstd::Encoder::new(writer, level).map_err(|e| Error::io(e, "creating zstd encoder"))?;
     zstd_encoder
         .include_checksum(true)
         .map_err(|e| Error::io(e, "setting zstd checksum"))?;
@@ -213,10 +220,18 @@ fn zstd_encoder<'a, W: io::Write>(writer: W) -> Result<zstd::stream::AutoFinishE
 #[cfg(feature = "async")]
 fn async_zstd_encoder<W: tokio::io::AsyncWriteExt + Unpin>(
     writer: W,
+) -> async_compression::tokio::write::ZstdEncoder<W> {
+    async_zstd_encoder_with_clevel(writer, ZSTD_COMPRESSION_LEVEL)
+}
+
+#[cfg(feature = "async")]
+fn async_zstd_encoder_with_clevel<W: tokio::io::AsyncWriteExt + Unpin>(
+    writer: W,
+    level: i32,
 ) -> async_compression::tokio::write::ZstdEncoder<W> {
     async_compression::tokio::write::ZstdEncoder::with_quality_and_params(
         writer,
-        async_compression::Level::Precise(ZSTD_COMPRESSION_LEVEL),
+        async_compression::Level::Precise(level),
         &[async_compression::zstd::CParameter::checksum_flag(true)],
     )
 }

rust/dbn/src/encode.rs

@@ -1,6 +1,6 @@
 use std::io;
 
-use super::zstd_encoder;
+use super::{zstd_encoder, zstd_encoder_with_clevel};
 use crate::{Compression, Result};
 
 /// Type for runtime polymorphism over whether encoding uncompressed or ZStd-compressed
@@ -21,17 +21,27 @@ impl<W> DynWriter<'_, W>
 where
     W: io::Write,
 {
-    /// Create a new instance of [`DynWriter`] which will wrap `writer` with `compression`.
+    /// Creates a new instance of [`DynWriter`] which will wrap `writer` with `compression`.
     ///
     /// # Errors
-    /// This function returns an error if it fails to initialize the Zstd compression.
+    /// This function returns an error if it fails to initialize the Zstd encoder.
     pub fn new(writer: W, compression: Compression) -> Result<Self> {
         match compression {
             Compression::None => Ok(Self(DynWriterImpl::Uncompressed(writer))),
             Compression::ZStd => zstd_encoder(writer).map(|enc| Self(DynWriterImpl::Zstd(enc))),
         }
     }
 
+    /// Creates a new instance with zstd compression of the specified level.
+    ///
+    /// # Errors
+    /// This function returns an error if it fails to initialize the Zstd encoder.
+    pub fn with_compression_level(writer: W, level: i32) -> Result<Self> {
+        Ok(Self(DynWriterImpl::Zstd(zstd_encoder_with_clevel(
+            writer, level,
+        )?)))
+    }
+
     /// Returns a mutable reference to the underlying writer.
     pub fn get_mut(&mut self) -> &mut W {
         match &mut self.0 {
@@ -96,50 +106,40 @@ mod r#async {
     use async_compression::tokio::write::ZstdEncoder;
     use tokio::io::{self, BufWriter};
 
-    use crate::{encode::async_zstd_encoder, enums::Compression};
+    use crate::{
+        encode::{async_zstd_encoder, async_zstd_encoder_with_clevel},
+        enums::Compression,
+    };
 
     /// An object that allows for abstracting over compressed and uncompressed output
     /// with buffering.
-    pub struct DynBufWriter<W, B = W>(DynBufWriterImpl<W, B>)
-    where
-        W: io::AsyncWriteExt + Unpin,
-        B: io::AsyncWriteExt + Unpin;
+    pub struct DynBufWriter<W, B = W>(DynBufWriterImpl<W, B>);
 
-    enum DynBufWriterImpl<W, B>
-    where
-        W: io::AsyncWriteExt + Unpin,
-        B: io::AsyncWriteExt + Unpin,
-    {
+    enum DynBufWriterImpl<W, B> {
         Uncompressed(B),
         Zstd(ZstdEncoder<W>),
     }
 
-    impl<W> DynBufWriter<W>
+    impl<W> DynBufWriter<W, BufWriter<W>>
     where
         W: io::AsyncWriteExt + Unpin,
     {
-        /// Creates a new instance of [`DynWriter`] which will wrap `writer` with
+        /// Creates a new instance which will wrap `writer` in a `BufWriter` and
         /// `compression`.
         pub fn new(writer: W, compression: Compression) -> Self {
             Self(match compression {
-                Compression::None => DynBufWriterImpl::Uncompressed(writer),
+                Compression::None => DynBufWriterImpl::Uncompressed(BufWriter::new(writer)),
+                // async zstd always wraps the writer in a BufWriter
                 Compression::ZStd => DynBufWriterImpl::Zstd(async_zstd_encoder(writer)),
             })
         }
-    }
 
-    impl<W> DynBufWriter<W, BufWriter<W>>
-    where
-        W: io::AsyncWriteExt + Unpin,
-    {
-        /// Creates a new instance of [`DynWriter`], wrapping `writer` in a `BufWriter`.
-        pub fn new_buffered(writer: W, compression: Compression) -> Self {
-            Self(match compression {
-                Compression::None => DynBufWriterImpl::Uncompressed(BufWriter::new(writer)),
-                // `ZstdEncoder` already wraps `W` in a `BufWriter`, cf.
-                // https://github.com/Nullus157/async-compression/blob/main/src/tokio/write/generic/encoder.rs
-                Compression::ZStd => DynBufWriterImpl::Zstd(async_zstd_encoder(writer)),
-            })
+        /// Creates a new instance, wrapping `writer` in a `BufWriter` and compressing
+        /// the output according to `level`.
+        pub fn with_compression_level(writer: W, level: i32) -> Self {
+            Self(DynBufWriterImpl::Zstd(async_zstd_encoder_with_clevel(
+                writer, level,
+            )))
         }
     }
 
@@ -204,6 +204,13 @@ mod r#async {
             })
         }
 
+        /// Creates a new instance, compressing the output according to `level`.
+        pub fn with_compression_level(writer: W, level: i32) -> Self {
+            Self(DynWriterImpl::Zstd(async_zstd_encoder_with_clevel(
+                writer, level,
+            )))
+        }
+
         /// Returns a mutable reference to the underlying writer.
         pub fn get_mut(&mut self) -> &mut W {
             match &mut self.0 {

rust/dbn/src/encode/dyn_writer.rs

@@ -567,8 +567,7 @@ impl InstrumentDefMsg {
         c_chars_to_str(&self.cfi)
     }
 
-    /// Returns the type of the strument, e.g. FUT for future or future spread as
-    /// a `&str`.
+    /// Returns the [Security type](https://databento.com/docs/schemas-and-data-formats/instrument-definitions#security-type) of the instrument, e.g. FUT for future or future spread as a `&str`.
     ///
     /// # Errors
     /// This function returns an error if `security_type` contains invalid UTF-8.