[READ-ONLY] Mirror of https://github.com/probablykasper/csv-pipeline.
Improve errors
+70
-75
+13
-9
src/lib.rs
+13
-9
src/lib.rs
···
110
110
/// Alias of [`csv::StringRecord`]
111
111
pub type Row = csv::StringRecord;
112
112
/// Alias of `Result<Row, Error>`
113
-
pub type RowResult = Result<Row, Error>;
113
+
pub type RowResult = Result<Row, PlError>;
114
114
115
+
/// Error originating from the specified pipeline source index
115
116
#[derive(Debug)]
116
-
pub struct Error {
117
+
pub struct PlError {
118
+
pub error: Error,
117
119
pub source: usize,
118
-
pub kind: ErrorKind,
119
-
}
120
-
impl Error {
121
-
pub fn new(source: usize, kind: ErrorKind) -> Error {
122
-
Error { source, kind }
123
-
}
124
120
}
125
121
126
122
#[derive(Debug)]
127
-
pub enum ErrorKind {
123
+
pub enum Error {
128
124
/// CSV and IO errors are in here.
129
125
Csv(csv::Error),
130
126
/// The column of this name is missing.
···
136
132
/// Two pipeline sources don't have the same headers.
137
133
MismatchedHeaders(Row, Row),
138
134
}
135
+
impl Error {
136
+
pub fn at_source(self, source: usize) -> PlError {
137
+
PlError {
138
+
error: self,
139
+
source,
140
+
}
141
+
}
142
+
}
+14
-11
src/pipeline.rs
+14
-11
src/pipeline.rs
···
4
4
};
5
5
use crate::target::{StringTarget, Target};
6
6
use crate::transform::Transform;
7
-
use crate::{Error, ErrorKind, Row, RowResult};
7
+
use crate::{Error, PlError, Row, RowResult};
8
8
use csv::{Reader, ReaderBuilder, StringRecordsIntoIter};
9
9
use linked_hash_map::LinkedHashMap;
10
10
use std::borrow::BorrowMut;
···
19
19
}
20
20
21
21
impl<'a> Pipeline<'a> {
22
-
pub fn from_reader<R: io::Read + 'a>(mut reader: Reader<R>) -> Result<Self, Error> {
22
+
pub fn from_reader<R: io::Read + 'a>(mut reader: Reader<R>) -> Result<Self, PlError> {
23
23
let headers_row = reader.headers().unwrap().clone();
24
24
let row_iterator = RowIter::from_records(0, reader.into_records());
25
25
Ok(Pipeline {
26
26
headers: match Headers::from_row(headers_row) {
27
27
Ok(headers) => headers,
28
28
Err(duplicated_col) => {
29
-
return Err(Error::new(0, ErrorKind::DuplicateColumn(duplicated_col)))
29
+
return Err(Error::DuplicateColumn(duplicated_col).at_source(0))
30
30
}
31
31
},
32
32
source: 0,
···
35
35
}
36
36
37
37
/// Create a pipeline from a CSV or TSV file.
38
-
pub fn from_path<P: AsRef<Path>>(file_path: P) -> Result<Self, Error> {
38
+
pub fn from_path<P: AsRef<Path>>(file_path: P) -> Result<Self, PlError> {
39
39
let ext = file_path.as_ref().extension().unwrap_or_default();
40
40
let delimiter = match ext.to_string_lossy().as_ref() {
41
41
"tsv" => b'\t',
···
109
109
self.iterator = Box::new(AddCol {
110
110
iterator: self.iterator,
111
111
f: get_value,
112
+
source: self.source,
112
113
headers: self.headers.clone(),
113
114
});
114
115
self
···
139
140
self.iterator = Box::new(MapRow {
140
141
iterator: self.iterator,
141
142
f: get_row,
143
+
source: self.source,
142
144
headers: self.headers.clone(),
143
145
});
144
146
self
···
311
313
self.iterator = Box::new(Validate {
312
314
iterator: self.iterator,
313
315
f,
316
+
source: self.source,
314
317
headers: self.headers.clone(),
315
318
});
316
319
self
···
362
365
}
363
366
364
367
/// Shorthand for `.build().run()`.
365
-
pub fn run(self) -> Result<(), Error> {
368
+
pub fn run(self) -> Result<(), PlError> {
366
369
self.build().run()
367
370
}
368
371
369
-
pub fn collect_into_string(self) -> Result<String, Error> {
372
+
pub fn collect_into_string(self) -> Result<String, PlError> {
370
373
let mut csv = String::new();
371
374
self.flush(StringTarget::new(&mut csv)).run()?;
372
375
Ok(csv)
···
391
394
/// Advances the iterator until an error is found.
392
395
///
393
396
/// Returns `None` when the iterator is finished.
394
-
pub fn next_error(&mut self) -> Option<Error> {
397
+
pub fn next_error(&mut self) -> Option<PlError> {
395
398
while let Some(item) = self.next() {
396
399
if let Err(err) = item {
397
400
return Some(err);
···
401
404
}
402
405
403
406
/// Run through the whole iterator. Returns the first error found, if any
404
-
pub fn run(&mut self) -> Result<(), Error> {
407
+
pub fn run(&mut self) -> Result<(), PlError> {
405
408
while let Some(item) = self.next() {
406
409
item?;
407
410
}
···
434
437
fn next(&mut self) -> Option<Self::Item> {
435
438
self.inner.next().map(|result| {
436
439
result.map_err(|err| {
437
-
return Error::new(self.source, ErrorKind::Csv(err));
440
+
return Error::Csv(err).at_source(self.source);
438
441
})
439
442
})
440
443
}
···
451
454
.unwrap_err();
452
455
453
456
assert_eq!(err.source, 2);
454
-
match err.kind {
455
-
ErrorKind::MismatchedHeaders(h1, h2) => {
457
+
match err.error {
458
+
Error::MismatchedHeaders(h1, h2) => {
456
459
assert_eq!(h1, Row::from(vec!["A", "B"]));
457
460
assert_eq!(h2, Row::from(vec!["ID", "Country"]));
458
461
}
+28
-35
src/pipeline_iterators.rs
+28
-35
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, ErrorKind, Pipeline, PipelineIter, Row, RowResult};
4
+
use crate::{Error, Pipeline, PipelineIter, Row, RowResult};
5
5
use linked_hash_map::{Entry, LinkedHashMap};
6
6
7
7
pub struct PipelinesChain<'a, P> {
···
34
34
self.current = Some(pipeline.build());
35
35
let current = self.current.as_mut().unwrap();
36
36
if current.headers.get_row() != self.headers.get_row() {
37
-
return Some(Err(Error::new(
38
-
self.index,
39
-
ErrorKind::MismatchedHeaders(
40
-
self.headers.get_row().to_owned(),
41
-
current.headers.get_row().to_owned(),
42
-
),
43
-
)));
37
+
return Some(Err(Error::MismatchedHeaders(
38
+
self.headers.get_row().to_owned(),
39
+
current.headers.get_row().to_owned(),
40
+
)
41
+
.at_source(self.index)));
44
42
}
45
43
}
46
44
None => {
···
55
53
pub struct AddCol<I, F: FnMut(&Headers, &Row) -> Result<String, Error>> {
56
54
pub iterator: I,
57
55
pub f: F,
56
+
pub source: usize,
58
57
pub headers: Headers,
59
58
}
60
59
impl<I, F> Iterator for AddCol<I, F>
···
74
73
row.push_field(&value);
75
74
Some(Ok(row))
76
75
}
77
-
Err(e) => Some(Err(e)),
76
+
Err(e) => Some(Err(e.at_source(self.source))),
78
77
}
79
78
}
80
79
}
···
82
81
pub struct MapRow<I, F: FnMut(&Headers, Row) -> Result<Row, Error>> {
83
82
pub iterator: I,
84
83
pub f: F,
84
+
pub source: usize,
85
85
pub headers: Headers,
86
86
}
87
87
impl<I, F> Iterator for MapRow<I, F>
···
98
98
};
99
99
match (self.f)(&self.headers, row) {
100
100
Ok(value) => Some(Ok(value)),
101
-
Err(e) => Some(Err(e)),
101
+
Err(e) => Some(Err(e.at_source(self.source))),
102
102
}
103
103
}
104
104
}
···
126
126
let index = match self.index {
127
127
Some(index) => index,
128
128
None => {
129
-
return Some(Err(Error::new(
130
-
self.source,
131
-
ErrorKind::MissingColumn(self.name.clone()),
132
-
)))
129
+
return Some(Err(
130
+
Error::MissingColumn(self.name.clone()).at_source(self.source)
131
+
))
133
132
}
134
133
};
135
134
let field = match row_vec.get_mut(index) {
136
135
Some(field) => field,
137
136
None => {
138
-
return Some(Err(Error::new(
139
-
self.source,
140
-
ErrorKind::MissingColumn(self.name.clone()),
141
-
)))
137
+
return Some(Err(
138
+
Error::MissingColumn(self.name.clone()).at_source(self.source)
139
+
))
142
140
}
143
141
};
144
142
let new_value = match (self.f)(field) {
145
143
Ok(value) => value,
146
-
Err(e) => return Some(Err(e)),
144
+
Err(e) => return Some(Err(e.at_source(self.source))),
147
145
};
148
146
*field = &new_value;
149
147
Some(Ok(row_vec.into()))
···
171
169
for col in &self.columns {
172
170
let field = match self.headers.get_field(&row, col) {
173
171
Some(field) => field,
174
-
None => {
175
-
return Some(Err(Error::new(
176
-
self.source,
177
-
ErrorKind::MissingColumn(col.clone()),
178
-
)))
179
-
}
172
+
None => return Some(Err(Error::MissingColumn(col.clone()).at_source(self.source))),
180
173
};
181
174
selection.push(field);
182
175
}
···
212
205
};
213
206
let hash = match compute_hash(&self.hashers, &self.headers, &row) {
214
207
Ok(hash) => hash,
215
-
Err(e) => return Some(Err(Error::new(self.source, e))),
208
+
Err(e) => return Some(Err(e.at_source(self.source))),
216
209
};
217
210
218
211
match self.groups.entry(hash) {
···
227
220
for reducer in group_row {
228
221
let result = reducer.add_row(&self.headers, &row);
229
222
if let Err(e) = result {
230
-
return Some(Err(Error::new(self.source, e)));
223
+
return Some(Err(e.at_source(self.source)));
231
224
}
232
225
}
233
226
}
···
246
239
pub struct Validate<I, F> {
247
240
pub iterator: I,
248
241
pub f: F,
242
+
pub source: usize,
249
243
pub headers: Headers,
250
244
}
251
245
impl<I, F> Iterator for Validate<I, F>
···
262
256
};
263
257
match (self.f)(&self.headers, &row) {
264
258
Ok(()) => Some(Ok(row)),
265
-
Err(e) => Some(Err(e)),
259
+
Err(e) => Some(Err(e.at_source(self.source))),
266
260
}
267
261
}
268
262
}
···
289
283
let field = match self.headers.get_field(&row, &self.name) {
290
284
Some(field) => field,
291
285
None => {
292
-
return Some(Err(Error::new(
293
-
self.source,
294
-
ErrorKind::MissingColumn(self.name.clone()),
295
-
)))
286
+
return Some(Err(
287
+
Error::MissingColumn(self.name.clone()).at_source(self.source)
288
+
))
296
289
}
297
290
};
298
291
match (self.f)(&field) {
299
292
Ok(()) => Some(Ok(row)),
300
-
Err(e) => Some(Err(e)),
293
+
Err(e) => Some(Err(e.at_source(self.source))),
301
294
}
302
295
}
303
296
}
···
330
323
if let Some(headers) = &self.headers {
331
324
match self.target.write_headers(headers) {
332
325
Ok(()) => self.headers = None,
333
-
Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))),
326
+
Err(e) => return Some(Err(Error::Csv(e).at_source(self.source))),
334
327
}
335
328
}
336
329
···
340
333
};
341
334
let r = match self.target.write_row(&row) {
342
335
Ok(()) => Some(Ok(row)),
343
-
Err(e) => return Some(Err(Error::new(self.source, ErrorKind::Csv(e)))),
336
+
Err(e) => return Some(Err(Error::Csv(e).at_source(self.source))),
344
337
};
345
338
r
346
339
}
+15
-20
src/transform.rs
+15
-20
src/transform.rs
···
1
-
use crate::{ErrorKind, Headers, Row};
1
+
use crate::{Error, Headers, Row};
2
2
use core::fmt::Display;
3
3
use std::collections::hash_map::DefaultHasher;
4
4
use std::hash::{Hash, Hasher};
···
13
13
_hasher: &mut DefaultHasher,
14
14
_headers: &Headers,
15
15
_row: &Row,
16
-
) -> Result<(), ErrorKind> {
16
+
) -> Result<(), Error> {
17
17
Ok(())
18
18
}
19
19
···
21
21
fn name(&self) -> String;
22
22
23
23
/// Combine the row with the value
24
-
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind>;
24
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error>;
25
25
26
26
/// Turn the current value to a string
27
27
fn value(&self) -> String;
···
66
66
/// Reduce the values from this column into a single value using a closure
67
67
pub fn reduce<'a, R, V>(self, reduce: R, init: V) -> Box<dyn Transform + 'a>
68
68
where
69
-
R: FnMut(V, &str) -> Result<V, ErrorKind> + 'a,
69
+
R: FnMut(V, &str) -> Result<V, Error> + 'a,
70
70
V: Display + Clone + 'a,
71
71
{
72
72
Box::new(Reduce {
···
84
84
value: String,
85
85
}
86
86
impl Transform for KeepUnique {
87
-
fn hash(
88
-
&self,
89
-
hasher: &mut DefaultHasher,
90
-
headers: &Headers,
91
-
row: &Row,
92
-
) -> Result<(), ErrorKind> {
87
+
fn hash(&self, hasher: &mut DefaultHasher, headers: &Headers, row: &Row) -> Result<(), Error> {
93
88
let field = headers
94
89
.get_field(row, &self.from_col)
95
-
.ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?;
90
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?;
96
91
field.hash(hasher);
97
92
Ok(())
98
93
}
···
101
96
self.name.clone()
102
97
}
103
98
104
-
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> {
99
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
105
100
self.value = headers
106
101
.get_field(row, &self.from_col)
107
-
.ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?
102
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?
108
103
.to_string();
109
104
Ok(())
110
105
}
···
118
113
transformers: &Vec<Box<dyn Transform + 'a>>,
119
114
headers: &Headers,
120
115
row: &Row,
121
-
) -> Result<u64, ErrorKind> {
116
+
) -> Result<u64, Error> {
122
117
let mut hasher = DefaultHasher::new();
123
118
for transformer in transformers {
124
119
let result = transformer.hash(&mut hasher, &headers, &row);
···
137
132
}
138
133
impl<F, V> Transform for Reduce<F, V>
139
134
where
140
-
F: FnMut(V, &str) -> Result<V, ErrorKind>,
135
+
F: FnMut(V, &str) -> Result<V, Error>,
141
136
V: Display + Clone,
142
137
{
143
-
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> {
138
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
144
139
let field = headers
145
140
.get_field(row, &self.from_col)
146
-
.ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?
141
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?
147
142
.to_string();
148
143
self.value = (self.reduce)(self.value.clone(), &field)?;
149
144
Ok(())
···
167
162
where
168
163
V: Display + AddAssign + FromStr + Clone,
169
164
{
170
-
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), ErrorKind> {
165
+
fn add_row(&mut self, headers: &Headers, row: &Row) -> Result<(), Error> {
171
166
let field = headers
172
167
.get_field(row, &self.from_col)
173
-
.ok_or(ErrorKind::MissingColumn(self.from_col.clone()))?
168
+
.ok_or(Error::MissingColumn(self.from_col.clone()))?
174
169
.to_string();
175
170
let new: V = match field.parse() {
176
171
Ok(v) => v,
177
-
Err(_) => return Err(ErrorKind::InvalidField(field)),
172
+
Err(_) => return Err(Error::InvalidField(field)),
178
173
};
179
174
self.value += new;
180
175
Ok(())