etl_core/config/
component.rs

1//! Opaque per-component configuration passthrough.
2
3use super::ConfigError;
4use serde::Deserialize;
5use serde::de::DeserializeOwned;
6
7/// An opaque component section: `{ <type_tag>: { ...connector config... } }`.
8///
9/// The framework never interprets the body — it records which component
10/// type the section selects (`kafka`, `clickhouse`, `memory`, ...) and hands
11/// the raw YAML to that component's factory, which deserializes it into its
12/// own typed config via [`deserialize_into`](Self::deserialize_into).
13///
14/// The nested-block shape (exactly one key) is what lets every typed struct
15/// in the tree keep `deny_unknown_fields` — a flattened shape would disable
16/// that check.
17#[derive(Debug, Clone, PartialEq)]
18pub struct ComponentConfig {
19    type_tag: String,
20    raw: serde_yaml::Value,
21    /// Where this section sits in the pipeline config (`source`, `sink`,
22    /// `deserializer`) — set after parsing, used to prefix error paths.
23    section: Option<&'static str>,
24}
25
26impl ComponentConfig {
27    /// Which component implementation this section selects.
28    #[must_use]
29    pub fn type_tag(&self) -> &str {
30        &self.type_tag
31    }
32
33    /// Deserialize the opaque body into the component's typed config.
34    ///
35    /// Errors carry the full dotted path from the pipeline config root,
36    /// e.g. `source.kafka.brokers: missing field \`brokers\``.
37    pub fn deserialize_into<T: DeserializeOwned>(&self) -> Result<T, ConfigError> {
38        serde_path_to_error::deserialize(self.raw.clone()).map_err(|e| {
39            let inner_path = e.path().to_string();
40            let mut context = self.prefix();
41            if inner_path != "." && !inner_path.is_empty() {
42                context.push('.');
43                context.push_str(&inner_path);
44            }
45            ConfigError::Component {
46                context,
47                message: e.into_inner().to_string(),
48            }
49        })
50    }
51
52    /// Dotted location of this component in the config, for error messages.
53    fn prefix(&self) -> String {
54        match self.section {
55            Some(section) => format!("{section}.{}", self.type_tag),
56            None => self.type_tag.clone(),
57        }
58    }
59
60    pub(super) fn set_section(&mut self, section: &'static str) {
61        self.section = Some(section);
62    }
63
64    /// Build a component config programmatically (primarily for tests and
65    /// `etl-test` pipelines that skip YAML).
66    ///
67    /// `raw` is the opaque connector body as a [`YamlValue`](super::YamlValue)
68    /// (an `etl-core` re-export of `serde_yaml::Value` — see its docs for the
69    /// dependency-policy exemption).
70    pub fn new(type_tag: impl Into<String>, raw: super::YamlValue) -> Self {
71        ComponentConfig {
72            type_tag: type_tag.into(),
73            raw,
74            section: None,
75        }
76    }
77}
78
79impl<'de> Deserialize<'de> for ComponentConfig {
80    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
81    where
82        D: serde::Deserializer<'de>,
83    {
84        use serde::de::Error as _;
85        let mapping = serde_yaml::Mapping::deserialize(deserializer)?;
86        if mapping.len() != 1 {
87            return Err(D::Error::custom(format!(
88                "a component section must be a single-key mapping selecting the \
89                 component type, e.g. `kafka: {{ ... }}` — found {} keys",
90                mapping.len()
91            )));
92        }
93        let (key, value) = mapping.into_iter().next().expect("len checked above");
94        let type_tag = key.as_str().ok_or_else(|| {
95            D::Error::custom("component type tag must be a string, e.g. `kafka:`")
96        })?;
97        Ok(ComponentConfig {
98            type_tag: type_tag.to_owned(),
99            raw: value,
100            section: None,
101        })
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    #[derive(Debug, PartialEq, Deserialize)]
110    #[serde(deny_unknown_fields)]
111    struct FakeKafkaConfig {
112        brokers: String,
113        topic: String,
114        #[serde(default)]
115        batch: Batch,
116    }
117
118    #[derive(Debug, PartialEq, Deserialize, Default)]
119    #[serde(deny_unknown_fields)]
120    struct Batch {
121        max_rows: Option<u64>,
122    }
123
124    fn parse(yaml: &str) -> Result<ComponentConfig, serde_yaml::Error> {
125        serde_yaml::from_str(yaml)
126    }
127
128    #[test]
129    fn parses_single_key_section_and_deserializes_body() {
130        let cc = parse("kafka:\n  brokers: k1:9092\n  topic: orders\n").unwrap();
131        assert_eq!(cc.type_tag(), "kafka");
132        let typed: FakeKafkaConfig = cc.deserialize_into().unwrap();
133        assert_eq!(typed.brokers, "k1:9092");
134        assert_eq!(typed.topic, "orders");
135    }
136
137    #[test]
138    fn rejects_zero_and_multiple_keys() {
139        let err = parse("{}").unwrap_err().to_string();
140        assert!(err.contains("single-key mapping"), "{err}");
141        assert!(err.contains("0 keys"), "{err}");
142
143        let err = parse("kafka: {}\nmemory: {}\n").unwrap_err().to_string();
144        assert!(err.contains("2 keys"), "{err}");
145    }
146
147    #[test]
148    fn rejects_non_string_tag() {
149        let err = parse("7: {}").unwrap_err().to_string();
150        assert!(err.contains("type tag must be a string"), "{err}");
151    }
152
153    #[test]
154    fn error_paths_include_section_tag_and_field() {
155        let mut cc = parse("kafka:\n  brokers: k1:9092\n").unwrap();
156        cc.set_section("source");
157        let err = cc.deserialize_into::<FakeKafkaConfig>().unwrap_err();
158        let text = err.to_string();
159        assert!(text.starts_with("source.kafka"), "{text}");
160        assert!(text.contains("topic"), "{text}");
161    }
162
163    #[test]
164    fn nested_error_paths_point_at_the_field() {
165        let mut cc =
166            parse("kafka:\n  brokers: b\n  topic: t\n  batch:\n    max_rows: lots\n").unwrap();
167        cc.set_section("source");
168        let err = cc.deserialize_into::<FakeKafkaConfig>().unwrap_err();
169        let text = err.to_string();
170        assert!(text.contains("source.kafka.batch.max_rows"), "{text}");
171    }
172
173    #[test]
174    fn unknown_field_in_component_body_is_rejected_with_path() {
175        let cc = parse("kafka:\n  brokers: b\n  topic: t\n  bogus: 1\n").unwrap();
176        let err = cc.deserialize_into::<FakeKafkaConfig>().unwrap_err();
177        assert!(err.to_string().contains("bogus"), "{err}");
178    }
179
180    #[test]
181    fn empty_body_deserializes_into_defaultable_types() {
182        #[derive(Debug, Deserialize, Default)]
183        struct Empty {}
184        let cc = parse("memory:\n").unwrap();
185        assert_eq!(cc.type_tag(), "memory");
186        // `memory:` with no body is a null value; struct with no required
187        // fields must accept it.
188        let _typed: Option<Empty> = cc.deserialize_into().unwrap();
189    }
190}
etl_core/config/component.rs

etl_core/config/
component.rs
