pipeflow 0.0.4

A lightweight, configuration-driven data pipeline framework
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//! Error types for the pipeline

use thiserror::Error;

/// Pipeline error types
#[derive(Debug, Error)]
pub enum Error {
    /// Configuration error
    #[error("Configuration error: {0}")]
    Config(String),

    /// Transform processing error
    #[error("Transform error: {0}")]
    Transform(String),

    /// Source error
    #[error("Source error: {0}")]
    Source(String),

    /// Sink error
    #[error("Sink error: {0}")]
    Sink(String),

    /// JSON serialization/deserialization error
    #[error("JSON error: {0}")]
    Json(#[from] serde_json::Error),

    /// YAML parsing error
    #[error("YAML error: {0}")]
    Yaml(#[from] serde_yaml::Error),

    /// IO error
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),

    /// HTTP error
    #[cfg(feature = "http-client")]
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),
}

impl Error {
    /// Create a configuration error
    pub fn config(msg: impl Into<String>) -> Self {
        Self::Config(msg.into())
    }

    /// Create a transform error
    pub fn transform(msg: impl Into<String>) -> Self {
        Self::Transform(msg.into())
    }

    /// Create a source error
    pub fn source(msg: impl Into<String>) -> Self {
        Self::Source(msg.into())
    }

    /// Create a sink error
    pub fn sink(msg: impl Into<String>) -> Self {
        Self::Sink(msg.into())
    }
}

/// Result type alias for pipeline operations
pub type Result<T> = std::result::Result<T, Error>;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_error_config() {
        let err = Error::config("invalid config");
        assert!(matches!(err, Error::Config(_)));
        assert!(err.to_string().contains("Configuration error"));
        assert!(err.to_string().contains("invalid config"));
    }

    #[test]
    fn test_error_transform() {
        let err = Error::transform("transform failed");
        assert!(matches!(err, Error::Transform(_)));
        assert!(err.to_string().contains("Transform error"));
    }

    #[test]
    fn test_error_source() {
        let err = Error::source("source error");
        assert!(matches!(err, Error::Source(_)));
        assert!(err.to_string().contains("Source error"));
    }

    #[test]
    fn test_error_sink() {
        let err = Error::sink("sink error");
        assert!(matches!(err, Error::Sink(_)));
        assert!(err.to_string().contains("Sink error"));
    }

    #[test]
    fn test_error_from_io() {
        let io_err = std::io::Error::new(std::io::ErrorKind::NotFound, "file not found");
        let err: Error = io_err.into();
        assert!(matches!(err, Error::Io(_)));
        assert!(err.to_string().contains("IO error"));
    }

    #[test]
    fn test_error_from_json() {
        let invalid_json = "{ invalid }";
        let json_err = serde_json::from_str::<serde_json::Value>(invalid_json).unwrap_err();
        let err: Error = json_err.into();
        assert!(matches!(err, Error::Json(_)));
    }

    #[test]
    fn test_error_from_yaml() {
        let invalid_yaml = "invalid: [yaml: content";
        let yaml_err = serde_yaml::from_str::<serde_yaml::Value>(invalid_yaml).unwrap_err();
        let err: Error = yaml_err.into();
        assert!(matches!(err, Error::Yaml(_)));
    }
}