faucet_source_azure_blob/
config.rs

1//! Azure Blob source configuration.
2
3use faucet_common_azure::{AzureConnection, AzureCredentials};
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Format of objects stored in the container.
9#[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
10#[serde(rename_all = "snake_case")]
11pub enum AzureFileFormat {
12    /// Each line in the object is a separate JSON record.
13    #[default]
14    JsonLines,
15    /// The entire object is a JSON array of records.
16    JsonArray,
17    /// Each object becomes a single record with `"key"` and `"content"` fields.
18    RawText,
19}
20
21/// Configuration for the Azure Blob source connector.
22#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
23pub struct AzureBlobSourceConfig {
24    /// Azure connection (container, account, credentials, endpoint, …).
25    #[serde(flatten)]
26    pub connection: AzureConnection,
27    /// Object name prefix filter. Ignored when `object_keys` is set.
28    pub prefix: Option<String>,
29    /// Explicit object names. When set, listing is skipped and `prefix`
30    /// is ignored.
31    pub object_keys: Option<Vec<String>>,
32    /// File format.
33    #[serde(default)]
34    pub file_format: AzureFileFormat,
35    /// Hard cap on the number of objects read (after listing).
36    pub max_objects: Option<usize>,
37    /// Maximum concurrent object reads (default: 10).
38    #[serde(default = "default_concurrency")]
39    pub concurrency: usize,
40    /// Records per emitted `StreamPage`. `batch_size = 0` is the "no batching"
41    /// sentinel and emits one page per object.
42    #[serde(default = "default_batch_size")]
43    pub batch_size: usize,
44    /// Compression codec applied to each downloaded object. Defaults to
45    /// [`CompressionConfig::Auto`](faucet_core::CompressionConfig::Auto) — the
46    /// codec is resolved per-object-key, so a single source can read a mix of
47    /// compressed and uncompressed objects. Requires the crate-local
48    /// `compression` feature.
49    #[cfg(feature = "compression")]
50    #[serde(default)]
51    pub compression: faucet_core::CompressionConfig,
52}
53
54fn default_batch_size() -> usize {
55    DEFAULT_BATCH_SIZE
56}
57fn default_concurrency() -> usize {
58    10
59}
60
61impl AzureBlobSourceConfig {
62    /// Create a new config for `container` with sensible defaults.
63    pub fn new(container: impl Into<String>) -> Self {
64        Self {
65            connection: AzureConnection::new(container),
66            prefix: None,
67            object_keys: None,
68            file_format: AzureFileFormat::default(),
69            max_objects: None,
70            concurrency: default_concurrency(),
71            batch_size: default_batch_size(),
72            #[cfg(feature = "compression")]
73            compression: faucet_core::CompressionConfig::default(),
74        }
75    }
76
77    /// Set the storage-account name.
78    pub fn account(mut self, account: impl Into<String>) -> Self {
79        self.connection = self.connection.account(account);
80        self
81    }
82
83    /// Set the credential source.
84    pub fn auth(mut self, creds: AzureCredentials) -> Self {
85        self.connection = self.connection.auth(creds);
86        self
87    }
88
89    /// Set a custom blob endpoint (emulator / sovereign cloud).
90    pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
91        self.connection = self.connection.endpoint(endpoint);
92        self
93    }
94
95    /// Permit plaintext HTTP (required for the Azurite emulator).
96    pub fn allow_http(mut self, allow: bool) -> Self {
97        self.connection = self.connection.allow_http(allow);
98        self
99    }
100
101    /// Target the Azurite emulator.
102    pub fn use_emulator(mut self, use_emulator: bool) -> Self {
103        self.connection = self.connection.use_emulator(use_emulator);
104        self
105    }
106
107    /// Filter objects by name prefix.
108    pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
109        self.prefix = Some(prefix.into());
110        self
111    }
112
113    /// Read an explicit set of object names (skips listing).
114    pub fn object_keys(mut self, keys: Vec<String>) -> Self {
115        self.object_keys = Some(keys);
116        self
117    }
118
119    /// Set the object file format.
120    pub fn file_format(mut self, format: AzureFileFormat) -> Self {
121        self.file_format = format;
122        self
123    }
124
125    /// Cap the number of objects read.
126    pub fn max_objects(mut self, max: usize) -> Self {
127        self.max_objects = Some(max);
128        self
129    }
130
131    /// Set the maximum concurrent object reads.
132    pub fn concurrency(mut self, concurrency: usize) -> Self {
133        self.concurrency = concurrency;
134        self
135    }
136
137    /// Set the records-per-`StreamPage` hint.
138    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
139        self.batch_size = batch_size;
140        self
141    }
142
143    /// Set the compression codec. Available only with the `compression` feature.
144    #[cfg(feature = "compression")]
145    pub fn compression(mut self, c: faucet_core::CompressionConfig) -> Self {
146        self.compression = c;
147        self
148    }
149
150    /// The container name.
151    pub fn container(&self) -> &str {
152        &self.connection.container
153    }
154}
155
156#[cfg(test)]
157mod tests {
158    use super::*;
159
160    #[test]
161    fn default_config() {
162        let config = AzureBlobSourceConfig::new("my-container");
163        assert_eq!(config.container(), "my-container");
164        assert!(config.prefix.is_none());
165        assert!(config.object_keys.is_none());
166        assert_eq!(config.connection.auth, AzureCredentials::Default);
167        assert_eq!(config.file_format, AzureFileFormat::JsonLines);
168        assert!(config.max_objects.is_none());
169        assert_eq!(config.concurrency, 10);
170        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
171    }
172
173    #[test]
174    fn builder_methods() {
175        let config = AzureBlobSourceConfig::new("c")
176            .account("acct")
177            .prefix("data/")
178            .file_format(AzureFileFormat::JsonArray)
179            .max_objects(5)
180            .concurrency(20)
181            .with_batch_size(250)
182            .auth(AzureCredentials::AccountKey {
183                account_key: "k".into(),
184            });
185
186        assert_eq!(config.container(), "c");
187        assert_eq!(config.connection.account.as_deref(), Some("acct"));
188        assert_eq!(config.prefix.as_deref(), Some("data/"));
189        assert_eq!(config.file_format, AzureFileFormat::JsonArray);
190        assert_eq!(config.max_objects, Some(5));
191        assert_eq!(config.concurrency, 20);
192        assert_eq!(config.batch_size, 250);
193        assert!(matches!(
194            config.connection.auth,
195            AzureCredentials::AccountKey { .. }
196        ));
197    }
198
199    #[test]
200    fn file_format_default_is_json_lines() {
201        assert_eq!(AzureFileFormat::default(), AzureFileFormat::JsonLines);
202    }
203
204    #[test]
205    fn deserializes_flattened_connection_and_auth() {
206        let json = r#"{
207            "container": "c",
208            "account": "acct",
209            "auth": { "type": "account_key", "config": { "account_key": "k" } },
210            "prefix": "data/",
211            "file_format": "json_array"
212        }"#;
213        let config: AzureBlobSourceConfig = serde_json::from_str(json).unwrap();
214        assert_eq!(config.container(), "c");
215        assert_eq!(config.connection.account.as_deref(), Some("acct"));
216        assert_eq!(config.file_format, AzureFileFormat::JsonArray);
217        assert!(matches!(
218            config.connection.auth,
219            AzureCredentials::AccountKey { .. }
220        ));
221        // batch_size / concurrency default when omitted.
222        assert_eq!(config.batch_size, faucet_core::DEFAULT_BATCH_SIZE);
223        assert_eq!(config.concurrency, 10);
224    }
225
226    #[test]
227    fn auth_defaults_to_default_when_absent_from_json() {
228        let json = r#"{ "container": "c" }"#;
229        let config: AzureBlobSourceConfig = serde_json::from_str(json).unwrap();
230        assert_eq!(config.connection.auth, AzureCredentials::Default);
231    }
232
233    #[test]
234    fn batch_size_zero_is_accepted_as_no_batching_sentinel() {
235        let config = AzureBlobSourceConfig::new("c").with_batch_size(0);
236        assert_eq!(config.batch_size, 0);
237        assert!(faucet_core::validate_batch_size(config.batch_size).is_ok());
238    }
239
240    #[test]
241    fn batch_size_above_max_is_rejected() {
242        let config =
243            AzureBlobSourceConfig::new("c").with_batch_size(faucet_core::MAX_BATCH_SIZE + 1);
244        assert!(faucet_core::validate_batch_size(config.batch_size).is_err());
245    }
246
247    #[test]
248    fn schema_generates_without_panicking() {
249        let _ = faucet_core::schema_for!(AzureBlobSourceConfig);
250    }
251
252    #[cfg(feature = "compression")]
253    #[test]
254    fn compression_default_is_auto() {
255        let cfg = AzureBlobSourceConfig::new("c");
256        assert_eq!(cfg.compression, faucet_core::CompressionConfig::Auto);
257    }
258}
faucet_source_azure_blob/config.rs

faucet_source_azure_blob/
config.rs
