apif_source_row/
csv.rs

1use crate::SourceReader;
2use crate::SourceRow;
3use anyhow::Result;
4use apif_source_error::SourceError;
5use std::io::{BufReader, Read, Seek};
6
7pub struct CsvReader<R> {
8    reader: csv::Reader<BufReader<R>>,
9    headers: Vec<String>,
10    _delimiter: u8,
11    row_number: usize,
12    finished: bool,
13}
14
15impl<R: Read> CsvReader<R> {
16    pub fn new(reader: BufReader<R>, delimiter: u8) -> Result<Self> {
17        let mut csv_reader = csv::ReaderBuilder::new()
18            .delimiter(delimiter)
19            .comment(Some(b'#'))
20            .has_headers(true)
21            .flexible(false)
22            .from_reader(reader);
23
24        let headers = csv_reader
25            .headers()
26            .map_err(|e| anyhow::anyhow!("failed to read CSV header: {e}"))?
27            .iter()
28            .map(|h| h.to_string())
29            .collect::<Vec<_>>();
30
31        if headers.is_empty() {
32            return Err(SourceError::EmptyFile("csv".into()).into());
33        }
34
35        let mut seen = std::collections::HashSet::new();
36        for col in &headers {
37            if !seen.insert(col.as_str()) {
38                return Err(SourceError::DuplicateColumn(col.clone()).into());
39            }
40        }
41
42        Ok(Self {
43            reader: csv_reader,
44            headers,
45            _delimiter: delimiter,
46            row_number: 1,
47            finished: false,
48        })
49    }
50}
51
52impl<R: Read + Send> SourceReader for CsvReader<R> {
53    fn next_row(&mut self) -> Result<Option<SourceRow>> {
54        if self.finished {
55            return Ok(None);
56        }
57
58        if let Some(result) = self.reader.records().next() {
59            self.row_number += 1;
60            let record = match result {
61                Ok(r) => r,
62                Err(e) => {
63                    return Err(anyhow::anyhow!("CSV error at row {}: {e}", self.row_number));
64                }
65            };
66
67            let values: Vec<String> = record.iter().map(|f| f.to_string()).collect();
68
69            if values.len() != self.headers.len() {
70                return Err(SourceError::FieldCountMismatch {
71                    row: self.row_number,
72                    fields: values.len(),
73                    expected: self.headers.len(),
74                }
75                .into());
76            }
77
78            return Ok(Some(SourceRow::new(&self.headers, values)));
79        }
80
81        self.finished = true;
82        Ok(None)
83    }
84
85    fn headers(&self) -> &[String] {
86        &self.headers
87    }
88
89    fn reset(&mut self) -> Result<()> {
90        Ok(())
91    }
92}
93
94impl<R: Read + Seek + Send> CsvReader<R> {
95    pub fn reset_seekable(&mut self) -> Result<()> {
96        // csv::Reader doesn't expose the inner writer,
97        // so we need to replace it entirely by seeking the raw reader
98        Err(anyhow::anyhow!(
99            "reset_seekable not supported with csv crate reader"
100        ))
101    }
102}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use std::io::Cursor;
108
109    fn cursor(data: &str) -> BufReader<Cursor<&str>> {
110        BufReader::new(Cursor::new(data))
111    }
112
113    #[test]
114    fn csv_reads_basic_rows() {
115        let data = "id,name,age\n1,Alice,30\n2,Bob,25\n";
116        let mut reader = CsvReader::new(cursor(data), b',').unwrap();
117        assert_eq!(reader.headers(), &["id", "name", "age"]);
118
119        let row1 = reader.next_row().unwrap().unwrap();
120        assert_eq!(row1.get("id"), Some("1"));
121        assert_eq!(row1.get("name"), Some("Alice"));
122        assert_eq!(row1.get("age"), Some("30"));
123
124        let row2 = reader.next_row().unwrap().unwrap();
125        assert_eq!(row2.get("name"), Some("Bob"));
126
127        assert!(reader.next_row().unwrap().is_none());
128    }
129
130    #[test]
131    fn csv_handles_quoted_fields() {
132        let data = "id,name\n1,\"Smith, John\"\n2,\"O\"\"Brien\"\n";
133        let mut reader = CsvReader::new(cursor(data), b',').unwrap();
134
135        let row1 = reader.next_row().unwrap().unwrap();
136        assert_eq!(row1.get("name"), Some("Smith, John"));
137
138        let row2 = reader.next_row().unwrap().unwrap();
139        assert_eq!(row2.get("name"), Some("O\"Brien"));
140    }
141
142    #[test]
143    fn csv_skips_comments_and_blank_lines() {
144        let data = "id,val\n# comment\n\n1,hello\n\n2,world\n";
145        let mut reader = CsvReader::new(cursor(data), b',').unwrap();
146        let row1 = reader.next_row().unwrap().unwrap();
147        assert_eq!(row1.get("val"), Some("hello"));
148        let row2 = reader.next_row().unwrap().unwrap();
149        assert_eq!(row2.get("val"), Some("world"));
150        assert!(reader.next_row().unwrap().is_none());
151    }
152
153    #[test]
154    fn csv_empty_file_errors() {
155        let data = "";
156        let result = CsvReader::new(cursor(data), b',');
157        assert!(result.is_err());
158    }
159
160    #[test]
161    fn csv_duplicate_column_errors() {
162        let data = "id,id,value\n1,2,3\n";
163        let result = CsvReader::new(cursor(data), b',');
164        assert!(result.is_err());
165    }
166
167    #[test]
168    fn csv_field_count_mismatch_errors() {
169        let data = "id,name\n1,Alice,extra\n";
170        let mut reader = CsvReader::new(cursor(data), b',').unwrap();
171        let result = reader.next_row();
172        assert!(result.is_err());
173    }
174
175    #[test]
176    fn csv_custom_delimiter() {
177        let data = "id;name;age\n1;Alice;30\n";
178        let mut reader = CsvReader::new(cursor(data), b';').unwrap();
179        let row = reader.next_row().unwrap().unwrap();
180        assert_eq!(row.get("name"), Some("Alice"));
181    }
182}
apif_source_row/csv.rs

apif_source_row/
csv.rs
