Transform · kasper.space/csv-pipeline@eccd255

+8

src/lib.rs

··· 2 2 mod pipeline; 3 3 mod pipeline_iterators; 4 4 mod target; 5 + mod transform; 5 6 6 7 pub use headers::Headers; 7 8 pub use pipeline::{Pipeline, PipelineIter}; 8 9 pub use target::{PathTarget, StderrTarget, StdoutTarget, StringTarget, Target}; 10 + pub use transform::{Transform, Transformer}; 11 + 9 12 pub type Row = csv::StringRecord; 10 13 pub type RowResult = Result<Row, Error>; 11 14 ··· 15 18 Io(std::io::Error), 16 19 MissingColumn(String), 17 20 DuplicateColumn(String), 21 + InvalidField(String), 18 22 } 19 23 impl From<csv::Error> for Error { 20 24 fn from(error: csv::Error) -> Error { ··· 38 42 _ => Ok("Unknown"), 39 43 } 40 44 }) 45 + .transform_into([ 46 + Transformer::new("ID").keep_unique(), 47 + Transformer::new("Country").sum(0), 48 + ]) 41 49 .rename_col("Country", "COUNTRY") 42 50 .map_col("COUNTRY", |id_str| Ok(id_str.to_uppercase())) 43 51 .collect_into_string()

+30 -22

src/pipeline.rs

··· 1 1 use super::headers::Headers; 2 2 use crate::pipeline_iterators::{AddCol, Flush, MapCol, MapRow}; 3 3 use crate::target::Target; 4 + use crate::transform::Transform; 4 5 use crate::{Error, Row, RowResult, StringTarget}; 5 6 use csv::{Reader, ReaderBuilder, StringRecordsIntoIter}; 6 7 use std::borrow::BorrowMut; ··· 127 128 self 128 129 } 129 130 130 - /// Write to the specified [`Target`]. 131 - /// 132 - /// ## Example 133 - /// 134 - /// ``` 135 - /// use csv_pipeline::{Pipeline, StringTarget}; 136 - /// 137 - /// let mut csv = String::new(); 138 - /// Pipeline::from_path("test/AB.csv") 139 - /// .unwrap() 140 - /// .flush(StringTarget::new(&mut csv)) 141 - /// .run() 142 - /// .unwrap(); 143 - /// 144 - /// assert_eq!(csv, "A,B\n1,2\n"); 145 - /// ``` 146 - pub fn flush(mut self, target: impl Target + 'a) -> Self { 147 - let flush = Flush::new(self.iterator, target, self.headers.clone()); 148 - self.iterator = Box::new(flush); 149 - self 150 - } 151 - 152 131 /// Panics if a new name already exists 153 132 /// 154 133 /// ## Example ··· 205 184 } 206 185 } 207 186 self.headers = new_headers; 187 + self 188 + } 189 + 190 + pub fn transform_into<T>(self, transformers: T) -> Self 191 + where 192 + T: IntoIterator<Item = Box<dyn Transform>>, 193 + { 194 + self 195 + } 196 + 197 + /// Write to the specified [`Target`]. 198 + /// 199 + /// ## Example 200 + /// 201 + /// ``` 202 + /// use csv_pipeline::{Pipeline, StringTarget}; 203 + /// 204 + /// let mut csv = String::new(); 205 + /// Pipeline::from_path("test/AB.csv") 206 + /// .unwrap() 207 + /// .flush(StringTarget::new(&mut csv)) 208 + /// .run() 209 + /// .unwrap(); 210 + /// 211 + /// assert_eq!(csv, "A,B\n1,2\n"); 212 + /// ``` 213 + pub fn flush(mut self, target: impl Target + 'a) -> Self { 214 + let flush = Flush::new(self.iterator, target, self.headers.clone()); 215 + self.iterator = Box::new(flush); 208 216 self 209 217 } 210 218

+45 -2

src/pipeline_iterators.rs

··· 1 + use std::collections::hash_map::{DefaultHasher, Entry}; 2 + use std::collections::HashMap; 3 + use std::hash::Hasher; 4 + 1 5 use super::headers::Headers; 2 6 use crate::target::Target; 3 - use crate::{Error, Row, RowResult}; 7 + use crate::transform::compute_hash; 8 + use crate::{Error, Row, RowResult, Transform}; 4 9 5 10 pub struct AddCol<'a, I, F: FnMut(&Headers, &Row) -> Result<&'a str, Error>> { 6 11 pub iterator: I, ··· 89 94 } 90 95 } 91 96 97 + pub struct TransformInto<I> { 98 + pub iterator: I, 99 + pub groups: HashMap<u64, Vec<Box<dyn Transform>>>, 100 + pub transformers: Vec<Box<dyn Transform>>, 101 + pub headers: Headers, 102 + } 103 + impl<I> Iterator for TransformInto<I> 104 + where 105 + I: Iterator<Item = RowResult>, 106 + { 107 + type Item = RowResult; 108 + 109 + fn next(&mut self) -> Option<Self::Item> { 110 + // First run iterator into hashmap, then return rows from the hashmap 111 + // If any error rows are found, they are returned first 112 + if let Some(row_result) = self.iterator.next() { 113 + let row = match row_result { 114 + Ok(row) => row, 115 + Err(e) => return Some(Err(e)), 116 + }; 117 + let hash = match compute_hash(&self.transformers, &self.headers, &row) { 118 + Ok(hash) => hash, 119 + Err(e) => return Some(Err(e)), 120 + }; 121 + let group_row = self.groups.entry(hash).or_default(); 122 + for transformer in self.transformers { 123 + let result = transformer.add_row(&self.headers, &row); 124 + if let Err(e) = result { 125 + return Some(Err(e)); 126 + } 127 + } 128 + group_row.push_field(row); 129 + } else { 130 + } 131 + 132 + x 133 + } 134 + } 135 + 92 136 pub struct Flush<I, T> { 93 137 pub iterator: I, 94 138 pub target: T, 95 139 /// `None` if headers have been written, `Some` otherwise 96 140 headers: Option<Headers>, 97 141 } 98 - 99 142 impl<I, T> Flush<I, T> { 100 143 pub fn new(iterator: I, target: T, headers: Headers) -> Self { 101 144 Self {

+117

src/transform.rs

··· 1 + use crate::{Error, Headers, Row}; 2 + use std::collections::hash_map::DefaultHasher; 3 + use std::hash::{Hash, Hasher}; 4 + use std::ops::AddAssign; 5 + use std::str::FromStr; 6 + 7 + pub trait Transform { 8 + /// Add the row to the hasher to group this row separately from others 9 + fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { 10 + Ok(()) 11 + } 12 + 13 + /// Combine the row with the value 14 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error>; 15 + 16 + /// Turn the current value to a string 17 + fn value(&self) -> String; 18 + 19 + /// Get the resulting column name 20 + fn name(&self) -> String; 21 + } 22 + 23 + pub struct Transformer { 24 + col_name: String, 25 + from_col: String, 26 + } 27 + impl Transformer { 28 + pub fn new(col_name: &str) -> Self { 29 + Self { 30 + col_name: col_name.to_string(), 31 + from_col: col_name.to_string(), 32 + } 33 + } 34 + pub fn keep_unique(self) -> Box<dyn Transform> { 35 + Box::new(KeepUnique { 36 + col_name: self.col_name, 37 + from_col: self.from_col, 38 + value: "".to_string(), 39 + }) 40 + } 41 + pub fn sum<'a, N: AddAssign + FromStr + ToString + 'a>( 42 + self, 43 + init: N, 44 + ) -> Box<dyn Transform + 'a> { 45 + Box::new(Sum { 46 + col_name: self.col_name, 47 + from_col: self.from_col, 48 + value: init, 49 + }) 50 + } 51 + } 52 + 53 + pub fn compute_hash( 54 + transformers: &[Box<dyn Transform>], 55 + headers: &Headers, 56 + row: &Row, 57 + ) -> Result<u64, Error> { 58 + let mut hasher = DefaultHasher::new(); 59 + for transformer in transformers { 60 + let result = transformer.hash(&mut hasher, &headers, &row); 61 + if let Err(e) = result { 62 + return Err(e); 63 + } 64 + } 65 + Ok(hasher.finish()) 66 + } 67 + 68 + pub struct KeepUnique { 69 + col_name: String, 70 + from_col: String, 71 + value: String, 72 + } 73 + impl Transform for KeepUnique { 74 + fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> { 75 + let field = headers 76 + .get_field(row, &self.from_col) 77 + .ok_or(Error::MissingColumn(self.value.clone()))?; 78 + field.hash(hasher); 79 + Ok(()) 80 + } 81 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 82 + self.value = headers 83 + .get_field(row, &self.from_col) 84 + .ok_or(Error::MissingColumn(self.value.clone()))? 85 + .to_string(); 86 + Ok(()) 87 + } 88 + fn value(&self) -> String { 89 + self.value.clone() 90 + } 91 + fn name(&self) -> String { 92 + self.col_name.clone() 93 + } 94 + } 95 + 96 + pub struct Sum<N> { 97 + col_name: String, 98 + from_col: String, 99 + value: N, 100 + } 101 + impl<N: AddAssign + FromStr + ToString> Transform for Sum<N> { 102 + fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> { 103 + let field = headers.get_field(row, &self.col_name).unwrap(); 104 + let number = match N::from_str(field) { 105 + Ok(number) => number, 106 + Err(e) => return Err(Error::InvalidField(field.to_string())), 107 + }; 108 + self.value += number; 109 + Ok(()) 110 + } 111 + fn value(&self) -> String { 112 + self.value.to_string() 113 + } 114 + fn name(&self) -> String { 115 + self.col_name.clone() 116 + } 117 + }

Configure Feed

Configure Feed