dpcs 0.10.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

//! Quality gate model (SPEC Ch 12).

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

use super::{ExtensionMap, Metadata};

/// Quality gate attached to a pipeline, step, or interface.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "jsonschema", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
pub struct QualityGate {
    /// Stable quality-gate identifier.
    pub id: String,
    /// Logical purpose of the gate.
    pub purpose: String,
    /// One or more declarative evaluation criteria.
    pub criteria: Vec<QualityCriterion>,
    /// Behavior on successful evaluation.
    pub on_success: GateOutcome,
    /// Behavior on unsuccessful evaluation.
    pub on_failure: GateOutcome,
    /// Optional gate category.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub category: Option<String>,
    /// Placement of the gate within the pipeline.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub placement: Option<GatePlacement>,
    /// Optional descriptive metadata.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<Metadata>,
    /// Extension fields.
    #[serde(default, flatten)]
    pub extensions: ExtensionMap,
}

/// Declarative evaluation criterion for a quality gate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "jsonschema", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
pub struct QualityCriterion {
    /// Optional criterion identifier.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    /// Criterion type (for example `odcs`, `dtcs`, `expression`).
    #[serde(default, skip_serializing_if = "Option::is_none", rename = "type")]
    pub criterion_type: Option<String>,
    /// Reference to a contract in `contractReferences`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub contract_ref: Option<String>,
    /// Declarative expression or rule identifier.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expression: Option<String>,
    /// Extension fields.
    #[serde(default, flatten)]
    pub extensions: ExtensionMap,
}

/// Declared outcome of gate evaluation.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GateOutcome {
    /// Abort pipeline execution.
    Abort,
    /// Retry evaluation or upstream work.
    Retry,
    /// Route to an alternate path.
    AlternatePath,
    /// Request approval before continuing.
    RequireApproval,
    /// Record diagnostics and continue according to policy.
    EmitDiagnostics,
    /// Continue pipeline execution.
    Continue,
    /// Implementation-defined outcome.
    Extension(String),
}

impl GateOutcome {
    /// Returns the wire-form outcome string.
    pub fn as_str(&self) -> &str {
        match self {
            Self::Abort => "abort",
            Self::Retry => "retry",
            Self::AlternatePath => "alternatePath",
            Self::RequireApproval => "requireApproval",
            Self::EmitDiagnostics => "emitDiagnostics",
            Self::Continue => "continue",
            Self::Extension(value) => value.as_str(),
        }
    }
}

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

impl<'de> Deserialize<'de> for GateOutcome {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let value = String::deserialize(deserializer)?;
        Ok(match value.as_str() {
            "abort" => Self::Abort,
            "retry" => Self::Retry,
            "alternatePath" => Self::AlternatePath,
            "requireApproval" => Self::RequireApproval,
            "emitDiagnostics" => Self::EmitDiagnostics,
            "continue" => Self::Continue,
            other => Self::Extension(other.to_owned()),
        })
    }
}

#[cfg(feature = "jsonschema")]
impl schemars::JsonSchema for GateOutcome {
    fn schema_name() -> String {
        "GateOutcome".to_owned()
    }

    fn json_schema(gen: &mut schemars::gen::SchemaGenerator) -> schemars::schema::Schema {
        String::json_schema(gen)
    }
}

/// Placement of a quality gate within a pipeline.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "jsonschema", derive(schemars::JsonSchema))]
#[serde(rename_all = "camelCase")]
pub struct GatePlacement {
    /// Placement kind (`beforePipeline`, `beforeStep`, `afterStep`, `beforeCompletion`, or extension).
    pub kind: String,
    /// Target step identifier when kind is `beforeStep` or `afterStep`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub step_id: Option<String>,
}

impl GatePlacement {
    /// Returns whether this placement targets a step.
    pub fn targets_step(&self) -> bool {
        let kind = self.kind.trim();
        kind.eq_ignore_ascii_case("beforeStep") || kind.eq_ignore_ascii_case("afterStep")
    }
}