Add from_pipelines constructor · kasper.space/csv-pipeline@1980201

+4

CHANGELOG.md

··· 1 1 # Changelog 2 2 3 + ## Next 4 + - Add Pipeline select method 5 + - Add Pipeline `from_pipelines` constructor for merging pipelines together 6 + 3 7 ## 0.2.0 - 2023 Jan 11 4 8 - Add `Target` struct helper for creating targets, and hide the targets in the `target` module. 5 9 - Publish `Transform` trait

+2 -1

src/headers.rs

··· 16 16 } 17 17 } 18 18 19 - /// Returns `false` if `from` is non-existant or if the new name already exists 19 + /// Returns `Error::MissingColumn` if `from` is non-existant or `Error::DuplicateColumn` the new name already exists 20 20 pub fn rename(&mut self, from: &str, to: &str) -> Result<(), Error> { 21 21 if self.contains(to) { 22 22 return Err(Error::DuplicateColumn(to.to_string())); ··· 60 60 &self.row 61 61 } 62 62 63 + /// Returns `Error::DuplicateColumn` if a column is duplicated 63 64 pub fn from_row(row: Row) -> Result<Self, Error> { 64 65 let mut header = Headers::new(); 65 66 for field in &row {

+6

src/lib.rs

··· 114 114 115 115 #[derive(Debug)] 116 116 pub enum Error { 117 + /// cSV and IO errors are in here 117 118 Csv(csv::Error), 118 119 Io(std::io::Error), 120 + /// The column of this name is missing. 119 121 MissingColumn(String), 122 + /// This column name appears twice. 120 123 DuplicateColumn(String), 124 + /// This field has an invalid format. 121 125 InvalidField(String), 126 + /// Two pipeline sources don't have the same headers. 127 + MismatchedHeaders(Row, Row), 122 128 } 123 129 impl From<csv::Error> for Error { 124 130 fn from(error: csv::Error) -> Error {

+58 -5

src/pipeline.rs

··· 1 1 use super::headers::Headers; 2 2 use crate::pipeline_iterators::{ 3 - AddCol, Flush, MapCol, MapRow, Select, TransformInto, Validate, ValidateCol, 3 + AddCol, Flush, MapCol, MapRow, PipelinesChain, Select, TransformInto, Validate, ValidateCol, 4 4 }; 5 5 use crate::target::{StringTarget, Target}; 6 6 use crate::transform::Transform; ··· 42 42 Self::from_reader(reader) 43 43 } 44 44 45 + /// Merge multiple source pipelines into one. The source pipelines must have identical headers, otherwise the pipelie will return a [`MismatchedHeaders`](Error::MismatchedHeaders) error returned. 46 + /// 47 + /// ## Example 48 + /// 49 + /// ``` 50 + /// use csv_pipeline::Pipeline; 51 + /// 52 + /// let csv = Pipeline::from_pipelines(vec![ 53 + /// Pipeline::from_path("test/AB.csv").unwrap(), 54 + /// Pipeline::from_path("test/AB.csv").unwrap(), 55 + /// ]) 56 + /// .collect_into_string() 57 + /// .unwrap(); 58 + /// 59 + /// assert_eq!(csv, "A,B\n1,2\n1,2\n"); 60 + /// ``` 61 + pub fn from_pipelines(pipelines: Vec<Pipeline<'a>>) -> Self { 62 + let mut pipelines = pipelines.into_iter(); 63 + let current = pipelines.next(); 64 + let headers = match current { 65 + Some(ref pipeline) => pipeline.headers.clone(), 66 + None => Headers::new(), 67 + }; 68 + Pipeline { 69 + headers: headers.clone(), 70 + iterator: Box::new(PipelinesChain { 71 + pipelines, 72 + current: current.map(|p| p.build()), 73 + index: 0, 74 + headers, 75 + }), 76 + } 77 + } 78 + 45 79 /// Adds a column with values computed from the closure for each row. 46 80 /// 47 81 /// ## Example ··· 258 292 } 259 293 } 260 294 261 - pub fn validate<F>(mut self, get_row: F) -> Self 295 + pub fn validate<F>(mut self, f: F) -> Self 262 296 where 263 297 F: FnMut(&Headers, &Row) -> Result<(), Error> + 'a, 264 298 { 265 299 self.iterator = Box::new(Validate { 266 300 iterator: self.iterator, 267 - f: get_row, 301 + f, 268 302 headers: self.headers.clone(), 269 303 }); 270 304 self 271 305 } 272 306 273 - pub fn validate_col<F>(mut self, name: &str, get_row: F) -> Self 307 + pub fn validate_col<F>(mut self, name: &str, f: F) -> Self 274 308 where 275 309 F: FnMut(&str) -> Result<(), Error> + 'a, 276 310 { 277 311 self.iterator = Box::new(ValidateCol { 278 312 name: name.to_string(), 279 313 iterator: self.iterator, 280 - f: get_row, 314 + f, 281 315 headers: self.headers.clone(), 282 316 }); 283 317 self ··· 388 422 }) 389 423 } 390 424 } 425 + 426 + #[test] 427 + fn from_pipelines_mismatch() { 428 + let err = Pipeline::from_pipelines(vec![ 429 + Pipeline::from_path("test/AB.csv").unwrap(), 430 + Pipeline::from_path("test/AB.csv").unwrap(), 431 + Pipeline::from_path("test/Countries.csv").unwrap(), 432 + ]) 433 + .collect_into_string() 434 + .unwrap_err(); 435 + 436 + match err { 437 + Error::MismatchedHeaders(h1, h2) => { 438 + assert_eq!(h1, Row::from(vec!["A", "B"])); 439 + assert_eq!(h2, Row::from(vec!["ID", "Country"])); 440 + } 441 + _ => panic!("Expected MismatchedHeaders"), 442 + } 443 + }

+43 -1

src/pipeline_iterators.rs

··· 1 1 use super::headers::Headers; 2 2 use crate::target::Target; 3 3 use crate::transform::{compute_hash, Transform}; 4 - use crate::{Error, Row, RowResult}; 4 + use crate::{Error, Pipeline, PipelineIter, Row, RowResult}; 5 5 use std::collections::btree_map::Entry; 6 6 use std::collections::BTreeMap; 7 + 8 + pub struct PipelinesChain<'a, P> { 9 + pub pipelines: P, 10 + pub current: Option<PipelineIter<'a>>, 11 + pub index: usize, 12 + pub headers: Headers, 13 + } 14 + impl<'a, P> Iterator for PipelinesChain<'a, P> 15 + where 16 + P: Iterator<Item = Pipeline<'a>>, 17 + { 18 + type Item = RowResult; 19 + 20 + fn next(&mut self) -> Option<Self::Item> { 21 + match &mut self.current { 22 + Some(current) => match current.next() { 23 + Some(row) => return Some(row), 24 + None => {} 25 + }, 26 + None => {} 27 + }; 28 + match self.pipelines.next() { 29 + Some(pipeline) => { 30 + if pipeline.headers.get_row() != self.headers.get_row() { 31 + return Some(Err(Error::MismatchedHeaders( 32 + self.headers.get_row().to_owned(), 33 + pipeline.headers.get_row().to_owned(), 34 + ))); 35 + } 36 + self.current = Some(pipeline.build()); 37 + self.index += 1; 38 + } 39 + None => { 40 + self.current = None; 41 + } 42 + } 43 + match self.current { 44 + Some(ref mut current) => current.next(), 45 + None => None, 46 + } 47 + } 48 + } 7 49 8 50 pub struct AddCol<I, F: FnMut(&Headers, &Row) -> Result<String, Error>> { 9 51 pub iterator: I,

Configure Feed

Configure Feed