apif_source_row/
lib.rs

1pub mod analyzer;
2pub mod csv;
3pub mod definition;
4pub mod detect;
5pub mod driven;
6pub mod filter;
7pub mod index;
8pub mod index_builder;
9pub mod memory;
10pub mod ndjson;
11pub mod schema;
12pub mod tsv;
13
14pub use analyzer::{
15    IndexReason, IndexRequirement, SourceUsageAnalyzer, SourceUsagePlan, effective_source_name,
16};
17pub use apif_source_error::SourceError;
18pub use apif_twoq_cache::TwoQCache;
19pub use csv::CsvReader;
20pub use definition::{IndexMode, JoinType, SourceDefinition};
21pub use detect::{SourceFormat, detect_format};
22pub use driven::{
23    FallbackReason, FallbackType, RuntimeFallbackPolicy, SourceDrivenConfig, SourceFallbackEvent,
24};
25pub use filter::{FilterCondition, matches_all as matches_filter_all};
26pub use index::{BloomFilter, IndexEntry, IndexEntryV4, SourceIndex, XorFilter};
27pub use memory::InMemorySource;
28pub use ndjson::NdjsonReader;
29pub use tsv::TsvReader;
30
31use anyhow::Result;
32use apif_utils::FileUtils;
33use std::io::{BufRead, BufReader, Read};
34use std::path::Path;
35
36pub trait SourceReader: Send {
37    fn next_row(&mut self) -> Result<Option<SourceRow>>;
38    fn headers(&self) -> &[String];
39
40    /// Attempt to reset the reader to the beginning.
41    /// Returns `Ok(())` even if unsupported — check `supports_reset()` first.
42    fn reset(&mut self) -> Result<()>;
43
44    /// Whether `reset()` is actually supported by this reader.
45    /// Readers wrapping non-seekable streams (stdin, network) return false.
46    fn supports_reset(&self) -> bool {
47        false
48    }
49}
50
51pub fn open_source_reader(
52    definition: &SourceDefinition,
53    document_path: &Path,
54) -> Result<Box<dyn SourceReader>> {
55    let resolved = resolve_source_path(definition, document_path);
56    let file = std::fs::File::open(&resolved)
57        .map_err(|e| SourceError::FileOpenFailed(resolved.display().to_string(), e))?;
58
59    let format = if let Some(fmt) = &definition.format {
60        fmt.clone()
61    } else {
62        detect_format(&resolved)?
63    };
64
65    let reader = BufReader::new(file);
66    match format {
67        SourceFormat::Csv => {
68            let delimiter = definition.delimiter.unwrap_or(b',');
69            Ok(Box::new(CsvReader::new(reader, delimiter)?))
70        }
71        SourceFormat::Tsv => Ok(Box::new(TsvReader::new(reader)?)),
72        SourceFormat::Ndjson => Ok(Box::new(NdjsonReader::new(reader))),
73    }
74}
75
76pub fn resolve_source_path(
77    definition: &SourceDefinition,
78    document_path: &Path,
79) -> std::path::PathBuf {
80    FileUtils::resolve_relative_path(document_path, &definition.file)
81}
82
83pub fn peek_format(reader: &mut BufReader<impl Read>) -> Result<SourceFormat> {
84    let n = reader.fill_buf()?;
85    let text = String::from_utf8_lossy(n);
86    Ok(detect::detect_format_from_content(&text))
87}
88
89pub fn row_to_template_variables(
90    source_name: &str,
91    row: &SourceRow,
92) -> std::collections::HashMap<String, serde_json::Value> {
93    let mut vars = std::collections::HashMap::with_capacity(row.len() * 2);
94
95    for col in row.columns() {
96        if let Some(val) = row.get(col) {
97            let namespaced = format!("{source_name}.{col}");
98            vars.insert(namespaced, serde_json::Value::String(val.to_string()));
99        }
100    }
101
102    vars
103}
104
105#[cfg(test)]
106mod tests {
107    use crate::*;
108
109    #[test]
110    fn row_to_template_variables_namespaced() {
111        let row = SourceRow::from_pairs(vec![
112            ("id".into(), "42".into()),
113            ("name".into(), "Alice".into()),
114        ]);
115        let vars = row_to_template_variables("users", &row);
116
117        assert_eq!(
118            vars.get("users.id"),
119            Some(&serde_json::Value::String("42".into()))
120        );
121        assert_eq!(
122            vars.get("users.name"),
123            Some(&serde_json::Value::String("Alice".into()))
124        );
125        assert!(!vars.contains_key("id"));
126    }
127
128    #[test]
129    fn row_to_template_variables_empty_row() {
130        let row = SourceRow::from_pairs(vec![]);
131        let vars = row_to_template_variables("src", &row);
132        assert!(vars.is_empty());
133    }
134}
135
136// ── SourceRow ──────────────────────────────────────────────
137
138#[derive(Debug, Clone)]
139pub struct SourceRow {
140    columns: Vec<String>,
141    values: Vec<String>,
142}
143
144impl SourceRow {
145    pub fn new(headers: &[String], values: Vec<String>) -> Self {
146        Self {
147            columns: headers.to_vec(),
148            values,
149        }
150    }
151
152    pub fn from_csv_line(line: &str) -> Self {
153        let mut columns = Vec::new();
154        let mut values = Vec::new();
155        for part in line.split(',') {
156            let part = part.trim_ascii();
157            values.push(part.to_string());
158            if columns.len() < values.len() {
159                columns.push(format!("col_{}", columns.len()));
160            }
161        }
162        Self { columns, values }
163    }
164
165    pub fn from_pairs(pairs: Vec<(String, String)>) -> Self {
166        let mut columns = Vec::with_capacity(pairs.len());
167        let mut values = Vec::with_capacity(pairs.len());
168        for (k, v) in pairs {
169            columns.push(k);
170            values.push(v);
171        }
172        Self { columns, values }
173    }
174
175    pub fn get(&self, column: &str) -> Option<&str> {
176        let idx = self.columns.iter().position(|c| c == column)?;
177        self.values.get(idx).map(|s| s.as_str())
178    }
179
180    pub fn get_or(&self, column: &str, default: &str) -> String {
181        self.get(column)
182            .map(|s| s.to_string())
183            .unwrap_or_else(|| default.to_string())
184    }
185
186    pub fn columns(&self) -> &[String] {
187        &self.columns
188    }
189
190    pub fn values(&self) -> &[String] {
191        &self.values
192    }
193
194    pub fn to_map(&self) -> std::collections::HashMap<String, String> {
195        let mut map = std::collections::HashMap::with_capacity(self.columns.len());
196        for (i, col) in self.columns.iter().enumerate() {
197            if let Some(v) = self.values.get(i) {
198                map.insert(col.clone(), v.clone());
199            }
200        }
201        map
202    }
203
204    pub fn len(&self) -> usize {
205        self.values.len()
206    }
207
208    pub fn is_empty(&self) -> bool {
209        self.values.is_empty()
210    }
211}
212
213#[cfg(test)]
214mod source_row_tests {
215    use super::*;
216
217    #[test]
218    fn row_from_csv_line() {
219        let row = SourceRow::from_csv_line("1,Alice,engineer");
220        assert_eq!(row.columns(), &["col_0", "col_1", "col_2"]);
221        assert_eq!(row.values(), &["1", "Alice", "engineer"]);
222        assert_eq!(row.len(), 3);
223    }
224
225    #[test]
226    fn row_from_csv_empty_line() {
227        let row = SourceRow::from_csv_line("");
228        // split("") on empty string produces [""]
229        assert_eq!(row.len(), 1);
230        assert_eq!(row.get("col_0"), Some(""));
231    }
232
233    #[test]
234    fn row_new_and_get() {
235        let headers = vec!["id".into(), "name".into()];
236        let row = SourceRow::new(&headers, vec!["42".into(), "Alice".into()]);
237        assert_eq!(row.get("id"), Some("42"));
238        assert_eq!(row.get("name"), Some("Alice"));
239        assert_eq!(row.get("missing"), None);
240    }
241
242    #[test]
243    fn row_from_pairs() {
244        let row = SourceRow::from_pairs(vec![("x".into(), "1".into()), ("y".into(), "2".into())]);
245        assert_eq!(row.get("x"), Some("1"));
246        assert_eq!(row.get("y"), Some("2"));
247    }
248
249    #[test]
250    fn row_get_or_default() {
251        let headers = vec!["id".into()];
252        let row = SourceRow::new(&headers, vec!["1".into()]);
253        assert_eq!(row.get_or("id", "fallback"), "1");
254        assert_eq!(row.get_or("missing", "fallback"), "fallback");
255    }
256
257    #[test]
258    fn row_to_map() {
259        let headers = vec!["a".into(), "b".into()];
260        let row = SourceRow::new(&headers, vec!["1".into(), "2".into()]);
261        let map = row.to_map();
262        assert_eq!(map.get("a"), Some(&"1".to_string()));
263        assert_eq!(map.get("b"), Some(&"2".to_string()));
264    }
265
266    #[test]
267    fn row_len_and_empty() {
268        let empty_row = SourceRow::new(&[], vec![]);
269        assert!(empty_row.is_empty());
270        assert_eq!(empty_row.len(), 0);
271
272        let row = SourceRow::new(&["x".into()], vec!["1".into()]);
273        assert!(!row.is_empty());
274        assert_eq!(row.len(), 1);
275    }
276}
apif_source_row/lib.rs

apif_source_row/
lib.rs
