VER: Release 0.14.0

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 0.14.0 - 2024-10-01
+
+#### Enhancements
+- Made several previously internal functions public to allow advanced users more
+  customization and piecemeal usage of the live API:
+  - `ApiKey`
+  - `Symbols::to_chunked_api_string()`
+  - `live::protocol` module containing implementations of the raw API messages
+- Changed from `log` crate to `tracing` for better diagnostics
+- Upgraded DBN version to 0.22.0 for combined `_reserved3` and
+  `_reserved4` fields in `CbboMsg`
+
 ## 0.13.0 - 2024-09-25
 
 #### Enhancements

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "databento"
 authors = ["Databento <[email protected]>"]
-version = "0.13.0"
+version = "0.14.0"
 edition = "2021"
 repository = "https://github.com/databento/databento-rs"
 description = "Official Databento client library"
@@ -23,16 +23,12 @@ historical = ["dep:futures", "dep:reqwest", "dep:serde", "dep:tokio-util", "dep:
 live = ["dep:hex", "dep:sha2", "tokio/net"]
 
 [dependencies]
-# binary encoding
 dbn = { version = "0.21.0", features = ["async", "serde"] }
 # Async stream trait
 futures = { version = "0.3", optional = true }
 # Used for Live authentication
 hex = { version = "0.4", optional = true }
-log = "0.4"
-# HTTP client for historical API
 reqwest = { version = "0.12", optional = true, features = ["json", "stream"] }
-# JSON deserialization for historical API
 serde = { version = "1.0", optional = true, features = ["derive"] }
 serde_json = { version = "1.0", optional = true }
 # Used for Live authentication
@@ -42,12 +38,13 @@ time = { version = ">=0.3.35", features = ["macros", "parsing", "serde"] }
 tokio = { version = ">=1.28", features = ["io-util", "macros"] }
 # Stream utils
 tokio-util = { version = "0.7", features = ["io"], optional = true }
+tracing = "0.1"
 typed-builder = "0.20"
 
 [dev-dependencies]
 anyhow = "1.0.89"
 clap = { version = "4.5.18", features = ["derive"] }
-env_logger = "0.11.5"
 tempfile = "3.12.0"
 tokio = { version = "1.40", features = ["full"] }
+tracing-subscriber = "0.3.18"
 wiremock = "0.6"

src/historical/batch.rs

@@ -11,11 +11,11 @@ use std::{
 
 use dbn::{Compression, Encoding, SType, Schema};
 use futures::StreamExt;
-use log::info;
 use reqwest::RequestBuilder;
 use serde::{de, Deserialize, Deserializer};
 use time::OffsetDateTime;
 use tokio::io::BufWriter;
+use tracing::info;
 use typed_builder::TypedBuilder;
 
 use crate::{historical::check_http_error, Error, Symbols};
@@ -169,7 +169,7 @@ impl BatchClient<'_> {
             .map_err(|e| Error::internal(format!("Unable to parse URL: {e:?}")))?;
         let resp = self.inner.get_with_path(url.path())?.send().await?;
         let mut stream = check_http_error(resp).await?.bytes_stream();
-        info!("Saving {url} to {}", path.as_ref().display());
+        info!(%url, path=%path.as_ref().display(), "Downloading file");
         let mut output = BufWriter::new(
             tokio::fs::OpenOptions::new()
                 .create(true)

src/historical/client.rs

@@ -1,6 +1,6 @@
-use log::warn;
 use reqwest::{header::ACCEPT, IntoUrl, RequestBuilder, Url};
 use serde::Deserialize;
+use tracing::warn;
 
 use crate::{error::ApiError, ApiKey, Error};
 
@@ -207,7 +207,7 @@ fn check_warnings(response: &reqwest::Response) {
                 }
             }
             Err(err) => {
-                warn!("Failed to parse server warnings from HTTP header: {err:?}");
+                warn!(?err, "Failed to parse server warnings from HTTP header");
             }
         };
     };
@@ -263,7 +263,7 @@ impl ClientBuilder<Unset> {
     /// This function returns an error when the API key is invalid.
     pub fn key(self, key: impl ToString) -> crate::Result<ClientBuilder<ApiKey>> {
         Ok(ClientBuilder {
-            key: crate::validate_key(key.to_string())?,
+            key: ApiKey::new(key.to_string())?,
             base_url: self.base_url,
             gateway: self.gateway,
         })

src/lib.rs

@@ -35,9 +35,9 @@ pub use dbn;
 
 use std::fmt::{self, Display, Write};
 
-use log::error;
 #[cfg(feature = "historical")]
 use serde::{Deserialize, Deserializer};
+use tracing::error;
 
 /// A set of symbols for a particular [`SType`](dbn::enums::SType).
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -71,7 +71,9 @@ impl Symbols {
     }
 
     #[cfg(feature = "live")]
-    pub(crate) fn to_chunked_api_string(&self) -> Vec<String> {
+    /// Splits the symbol into chunks to stay within the message length requirements of
+    /// the live gateway.
+    pub fn to_chunked_api_string(&self) -> Vec<String> {
         const CHUNK_SIZE: usize = 128;
         match self {
             Symbols::All => vec![ALL_SYMBOLS.to_owned()],
@@ -160,31 +162,6 @@ impl From<Vec<&str>> for Symbols {
     }
 }
 
-pub(crate) fn validate_key(key: String) -> crate::Result<ApiKey> {
-    if key == "$YOUR_API_KEY" {
-        Err(Error::bad_arg(
-            "key",
-            "got placeholder API key '$YOUR_API_KEY'. Please pass a real API key",
-        ))
-    } else if key.len() != API_KEY_LENGTH {
-        Err(Error::bad_arg(
-            "key",
-            format!(
-                "expected to be 32-characters long, got {} characters",
-                key.len()
-            ),
-        ))
-    } else if !key.is_ascii() {
-        error!("API key '{key}' contains non-ASCII characters");
-        Err(Error::bad_arg(
-            "key",
-            "expected to be composed of only ASCII characters",
-        ))
-    } else {
-        Ok(ApiKey(key))
-    }
-}
-
 pub(crate) fn key_from_env() -> crate::Result<String> {
     std::env::var("DATABENTO_API_KEY").map_err(|e| {
         Error::bad_arg(
@@ -241,6 +218,49 @@ impl fmt::Debug for ApiKey {
     }
 }
 
+impl ApiKey {
+    /// Validates `key` meets requirements of an API key.
+    ///
+    /// # Errors
+    /// This function returns an error if the key is invalid.
+    pub fn new(key: String) -> crate::Result<ApiKey> {
+        if key == "$YOUR_API_KEY" {
+            Err(Error::bad_arg(
+                "key",
+                "got placeholder API key '$YOUR_API_KEY'. Please pass a real API key",
+            ))
+        } else if key.len() != API_KEY_LENGTH {
+            Err(Error::bad_arg(
+                "key",
+                format!(
+                    "expected to be 32-characters long, got {} characters",
+                    key.len()
+                ),
+            ))
+        } else if !key.is_ascii() {
+            error!("API key '{key}' contains non-ASCII characters");
+            Err(Error::bad_arg(
+                "key",
+                "expected to be composed of only ASCII characters",
+            ))
+        } else {
+            Ok(ApiKey(key))
+        }
+    }
+
+    /// Returns a slice of the last 5 characters of the key.
+    #[cfg(feature = "live")]
+    pub fn bucket_id(&self) -> &str {
+        // Safe to splice because validated as only containing ASCII characters in [`Self::new()`]
+        &self.0[API_KEY_LENGTH - BUCKET_ID_LENGTH..]
+    }
+
+    /// Returns the entire key as a slice.
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
+}
+
 #[cfg(test)]
 const TEST_DATA_PATH: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/data");
 #[cfg(test)]

src/live.rs

@@ -1,16 +1,18 @@
 //! The Live client and related API types. Used for both real-time data and intraday historical.
 
 mod client;
+pub mod protocol;
 
 use std::{net::SocketAddr, sync::Arc};
 
 use dbn::{SType, Schema, VersionUpgradePolicy};
-use log::warn;
 use time::{Duration, OffsetDateTime};
 use tokio::net::{lookup_host, ToSocketAddrs};
+use tracing::warn;
 use typed_builder::TypedBuilder;
 
 use crate::{ApiKey, Symbols};
+
 pub use client::Client;
 
 /// A subscription for real-time or intraday historical data.
@@ -132,7 +134,7 @@ impl<D> ClientBuilder<Unset, D> {
     pub fn key(self, key: impl ToString) -> crate::Result<ClientBuilder<ApiKey, D>> {
         Ok(ClientBuilder {
             addr: self.addr,
-            key: crate::validate_key(key.to_string())?,
+            key: ApiKey::new(key.to_string())?,
             dataset: self.dataset,
             send_ts_out: self.send_ts_out,
             upgrade_policy: self.upgrade_policy,