[READ-ONLY] Mirror of https://github.com/probablykasper/csv-pipeline.
Add code
+294
-3
+80
Cargo.lock
+80
Cargo.lock
···
1
+
# This file is automatically @generated by Cargo.
2
+
# It is not intended for manual editing.
3
+
version = 3
4
+
5
+
[[package]]
6
+
name = "bstr"
7
+
version = "0.2.17"
8
+
source = "registry+https://github.com/rust-lang/crates.io-index"
9
+
checksum = "ba3569f383e8f1598449f1a423e72e99569137b47740b1da11ef19af3d5c3223"
10
+
dependencies = [
11
+
"lazy_static",
12
+
"memchr",
13
+
"regex-automata",
14
+
"serde",
15
+
]
16
+
17
+
[[package]]
18
+
name = "csv"
19
+
version = "1.1.6"
20
+
source = "registry+https://github.com/rust-lang/crates.io-index"
21
+
checksum = "22813a6dc45b335f9bade10bf7271dc477e81113e89eb251a0bc2a8a81c536e1"
22
+
dependencies = [
23
+
"bstr",
24
+
"csv-core",
25
+
"itoa",
26
+
"ryu",
27
+
"serde",
28
+
]
29
+
30
+
[[package]]
31
+
name = "csv-core"
32
+
version = "0.1.10"
33
+
source = "registry+https://github.com/rust-lang/crates.io-index"
34
+
checksum = "2b2466559f260f48ad25fe6317b3c8dac77b5bdb5763ac7d9d6103530663bc90"
35
+
dependencies = [
36
+
"memchr",
37
+
]
38
+
39
+
[[package]]
40
+
name = "csv-pipeline"
41
+
version = "0.0.0"
42
+
dependencies = [
43
+
"csv",
44
+
]
45
+
46
+
[[package]]
47
+
name = "itoa"
48
+
version = "0.4.8"
49
+
source = "registry+https://github.com/rust-lang/crates.io-index"
50
+
checksum = "b71991ff56294aa922b450139ee08b3bfc70982c6b2c7562771375cf73542dd4"
51
+
52
+
[[package]]
53
+
name = "lazy_static"
54
+
version = "1.4.0"
55
+
source = "registry+https://github.com/rust-lang/crates.io-index"
56
+
checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
57
+
58
+
[[package]]
59
+
name = "memchr"
60
+
version = "2.5.0"
61
+
source = "registry+https://github.com/rust-lang/crates.io-index"
62
+
checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
63
+
64
+
[[package]]
65
+
name = "regex-automata"
66
+
version = "0.1.10"
67
+
source = "registry+https://github.com/rust-lang/crates.io-index"
68
+
checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132"
69
+
70
+
[[package]]
71
+
name = "ryu"
72
+
version = "1.0.12"
73
+
source = "registry+https://github.com/rust-lang/crates.io-index"
74
+
checksum = "7b4b9743ed687d4b4bcedf9ff5eaa7398495ae14e61cba0a295704edbc7decde"
75
+
76
+
[[package]]
77
+
name = "serde"
78
+
version = "1.0.152"
79
+
source = "registry+https://github.com/rust-lang/crates.io-index"
80
+
checksum = "bb7d1f0d3021d347a83e556fc4683dea2ea09d87bccdf88ff5c12545d89d5efb"
+2
-2
README.md
+2
-2
README.md
+1
rustfmt.toml
+1
rustfmt.toml
···
1
+
hard_tabs = true
+36
src/headers.rs
+36
src/headers.rs
···
1
+
use crate::Row;
2
+
use std::collections::HashMap;
3
+
4
+
#[derive(Debug, Clone, PartialEq)]
5
+
pub struct Headers {
6
+
indexes: HashMap<String, usize>,
7
+
names: Row,
8
+
}
9
+
impl Headers {
10
+
pub fn add(&mut self, name: &str) -> bool {
11
+
if self.indexes.contains_key(name) {
12
+
return false;
13
+
}
14
+
15
+
self.names.push_field(name);
16
+
self.indexes.insert(name.to_string(), self.names.len() - 1);
17
+
18
+
true
19
+
}
20
+
21
+
pub fn contains(&self, name: &str) -> bool {
22
+
self.indexes.contains_key(name)
23
+
}
24
+
}
25
+
impl From<Row> for Headers {
26
+
fn from(row: Row) -> Headers {
27
+
Headers {
28
+
indexes: row
29
+
.iter()
30
+
.enumerate()
31
+
.map(|(index, entry)| (entry.to_string(), index))
32
+
.collect(),
33
+
names: row,
34
+
}
35
+
}
36
+
}
+16
src/lib.rs
+16
src/lib.rs
···
1
+
pub mod headers;
2
+
pub mod pipe;
3
+
pub mod pipeline;
4
+
5
+
pub use headers::Headers;
6
+
pub use pipe::Pipe;
7
+
pub use pipeline::Pipeline;
8
+
9
+
#[derive(Debug, Clone, PartialEq)]
10
+
pub enum Error {
11
+
DuplicatedColumn(String),
12
+
Csv,
13
+
}
14
+
15
+
pub type Row = csv::StringRecord;
16
+
pub type RowResult = Result<Row, Error>;
+63
src/pipe.rs
+63
src/pipe.rs
···
1
+
use crate::RowResult;
2
+
3
+
pub type PipeIterator = Box<dyn Iterator<Item = RowResult>>;
4
+
5
+
pub struct Pipe {
6
+
iterator: PipeIterator,
7
+
}
8
+
impl Pipe {
9
+
pub fn new(iterator: PipeIterator) -> Self {
10
+
Self {
11
+
iterator: Box::new(iterator.into_iter()),
12
+
}
13
+
}
14
+
pub fn with_state<S>(self, state: S) -> StatefulPipeBuilder<S> {
15
+
StatefulPipeBuilder::new(self.iterator, state)
16
+
}
17
+
}
18
+
impl Iterator for Pipe {
19
+
type Item = RowResult;
20
+
21
+
fn next(&mut self) -> Option<Self::Item> {
22
+
self.iterator.next()
23
+
}
24
+
}
25
+
26
+
pub struct StatefulPipeBuilder<S> {
27
+
iterator: PipeIterator,
28
+
state: S,
29
+
}
30
+
impl<S> StatefulPipeBuilder<S> {
31
+
pub fn new(iterator: PipeIterator, state: S) -> Self {
32
+
Self { state, iterator }
33
+
}
34
+
pub fn map<F>(self, f: F) -> StatefulPipe<S, F>
35
+
where
36
+
F: FnMut(RowResult, &mut S) -> RowResult,
37
+
{
38
+
StatefulPipe {
39
+
iterator: self.iterator,
40
+
state: self.state,
41
+
f,
42
+
}
43
+
}
44
+
}
45
+
46
+
pub struct StatefulPipe<S, F: FnMut(RowResult, &mut S) -> RowResult> {
47
+
pub(crate) iterator: PipeIterator,
48
+
state: S,
49
+
f: F,
50
+
}
51
+
impl<S, F> Iterator for StatefulPipe<S, F>
52
+
where
53
+
F: FnMut(RowResult, &mut S) -> RowResult,
54
+
{
55
+
type Item = RowResult;
56
+
57
+
fn next(&mut self) -> Option<Self::Item> {
58
+
match self.iterator.next() {
59
+
Some(item) => Some((self.f)(item, &mut self.state)),
60
+
None => None,
61
+
}
62
+
}
63
+
}
+92
src/pipeline.rs
+92
src/pipeline.rs
···
1
+
use super::headers::Headers;
2
+
use super::pipe::{Pipe, PipeIterator};
3
+
use crate::{Error, Row, RowResult};
4
+
use csv::{Reader, ReaderBuilder};
5
+
use std::fs::File;
6
+
use std::path::Path;
7
+
8
+
pub struct Pipeline {
9
+
pub headers: Headers,
10
+
pipe: PipeIterator,
11
+
}
12
+
13
+
impl Pipeline {
14
+
pub fn from_reader(mut reader: Reader<File>) -> Self {
15
+
let headers_row = reader.headers().unwrap().clone();
16
+
let records = reader.into_records().map(|r| {
17
+
let row_result: RowResult = match r {
18
+
Ok(row) => Ok(row),
19
+
Err(err) => Err(Error::Csv),
20
+
};
21
+
row_result
22
+
});
23
+
Self {
24
+
headers: Headers::from(headers_row),
25
+
pipe: Box::new(records),
26
+
}
27
+
}
28
+
29
+
/// Create a pipeline from a CSV or TSV file.
30
+
pub fn from_path<P: AsRef<Path>>(file_path: P) -> Self {
31
+
let ext = file_path.as_ref().extension().unwrap_or_default();
32
+
let delimiter = match ext.to_string_lossy().as_ref() {
33
+
"tsv" => b'\t',
34
+
"csv" => b',',
35
+
_ => panic!("Unsupported file {}", file_path.as_ref().display()),
36
+
};
37
+
let reader = ReaderBuilder::new()
38
+
.delimiter(delimiter)
39
+
.from_path(file_path)
40
+
.unwrap();
41
+
Self::from_reader(reader)
42
+
}
43
+
44
+
/// Adds a column with values computed from the closure for each row.
45
+
///
46
+
/// ## Example
47
+
///
48
+
/// ```
49
+
/// use csv_pipeline::Pipeline;
50
+
///
51
+
/// Pipeline::from_path("test/Countries.csv")
52
+
/// .add_col("Language", |headers, row| {
53
+
/// Ok("".to_string())
54
+
/// });
55
+
/// ```
56
+
pub fn add_col<F>(mut self, name: &str, get_value: F) -> Self
57
+
where
58
+
F: FnMut(&Headers, &Row) -> Result<String, Error>,
59
+
{
60
+
self.headers.add(name);
61
+
62
+
struct State<F> {
63
+
get_value: F,
64
+
headers: Headers,
65
+
}
66
+
let pipe = Pipe::new(self.pipe).with_state(State {
67
+
get_value,
68
+
headers: self.headers.clone(),
69
+
});
70
+
let newpipe = pipe.map(|row_result, state| {
71
+
let mut row = row_result?;
72
+
let value = (state.get_value)(&state.headers, &row)?;
73
+
row.push_field(&value);
74
+
Ok(row)
75
+
});
76
+
77
+
self.pipe = Box::new(newpipe.iterator);
78
+
79
+
self
80
+
}
81
+
}
82
+
83
+
#[cfg(test)]
84
+
mod tests {
85
+
use crate::Pipeline;
86
+
87
+
#[test]
88
+
fn add_col() {
89
+
let mut pipeline = Pipeline::from_path("test/Countries.csv")
90
+
.add_col("Language", |_headers, _row| Ok("".to_string()));
91
+
}
92
+
}