dpcs 0.8.0

Reference implementation of the Data Pipeline Contract Standard (DPCS)
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156

//! Scheduling intent model (SPEC Ch 11).

use serde::{Deserialize, Deserializer, Serialize, Serializer};

use super::ExtensionMap;

/// Declared scheduling intent for a Pipeline Contract.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SchedulingIntent {
    /// Optional stable identifier for this intent declaration.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    /// Scheduling mode (required).
    pub mode: SchedulingMode,
    /// Optional cron-like schedule expression.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cron: Option<String>,
    /// Optional timezone identifier.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timezone: Option<String>,
    /// Execution frequency where applicable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub frequency: Option<String>,
    /// Execution windows.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub windows: Vec<String>,
    /// Blackout periods.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub blackouts: Vec<String>,
    /// Execution deadlines.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub deadlines: Vec<String>,
    /// Event declarations for event-driven modes.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub events: Vec<SchedulingEvent>,
    /// Timing and concurrency constraints.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub constraints: Option<SchedulingConstraints>,
    /// Applicable execution policies.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub policies: Vec<String>,
    /// Extension fields.
    #[serde(default, flatten)]
    pub extensions: ExtensionMap,
}

/// Scheduling mode for a Pipeline Contract.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SchedulingMode {
    /// Manual initiation.
    Manual,
    /// On-demand initiation.
    OnDemand,
    /// Time-based scheduled execution.
    Scheduled,
    /// Event-driven execution.
    EventDriven,
    /// Streaming execution intent.
    Streaming,
    /// Continuous execution intent.
    Continuous,
    /// Implementation-defined extension mode.
    Extension(String),
}

impl SchedulingMode {
    /// Returns the wire-form mode string.
    pub fn as_str(&self) -> &str {
        match self {
            Self::Manual => "manual",
            Self::OnDemand => "onDemand",
            Self::Scheduled => "scheduled",
            Self::EventDriven => "eventDriven",
            Self::Streaming => "streaming",
            Self::Continuous => "continuous",
            Self::Extension(value) => value.as_str(),
        }
    }

    /// Returns whether this is the scheduled (time-based) mode.
    pub fn is_scheduled(&self) -> bool {
        matches!(self, Self::Scheduled)
    }

    /// Returns whether this is the event-driven mode.
    pub fn is_event_driven(&self) -> bool {
        matches!(self, Self::EventDriven)
    }
}

impl Serialize for SchedulingMode {
    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        serializer.serialize_str(self.as_str())
    }
}

impl<'de> Deserialize<'de> for SchedulingMode {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let value = String::deserialize(deserializer)?;
        let normalized: String = value
            .chars()
            .filter(|ch| ch.is_ascii_alphanumeric())
            .flat_map(char::to_lowercase)
            .collect();
        Ok(match normalized.as_str() {
            "manual" => Self::Manual,
            "ondemand" => Self::OnDemand,
            "scheduled" => Self::Scheduled,
            "eventdriven" => Self::EventDriven,
            "streaming" => Self::Streaming,
            "continuous" => Self::Continuous,
            _ => Self::Extension(value),
        })
    }
}

/// Logical event that may initiate pipeline execution.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SchedulingEvent {
    /// Event identity.
    pub id: String,
    /// Event source.
    pub source: String,
    /// Triggering conditions.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub condition: Option<String>,
    /// Extension fields.
    #[serde(default, flatten)]
    pub extensions: ExtensionMap,
}

/// Timing and concurrency constraints for scheduling intent.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
#[serde(rename_all = "camelCase")]
pub struct SchedulingConstraints {
    /// Earliest execution time.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub earliest: Option<String>,
    /// Latest execution time.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub latest: Option<String>,
    /// Execution deadline.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub deadline: Option<String>,
    /// Concurrency limitations.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub concurrency: Option<String>,
    /// Ordering constraints.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ordering: Option<String>,
    /// Extension fields.
    #[serde(default, flatten)]
    pub extensions: ExtensionMap,
}