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

//! Dagster Definitions scaffold adapter.

use crate::diagnostics::ValidationReport;
use crate::plan::PipelinePlan;

use super::common::{python_file, semantics_file, PlanView};
use super::{BindContext, OrchestratorAdapter};
use crate::binding::artifact::{BindingFile, BindingTarget};

/// Dagster binding adapter.
pub struct DagsterAdapter;

impl OrchestratorAdapter for DagsterAdapter {
    fn target(&self) -> BindingTarget {
        BindingTarget::Dagster
    }

    fn translate(
        &self,
        plan: &PipelinePlan,
        ctx: &BindContext<'_>,
    ) -> Result<Vec<BindingFile>, ValidationReport> {
        let view = PlanView::new(plan, ctx);
        let idents = view.unique_python_idents(&[view.contract_id()]);
        let job_name = idents
            .get(view.contract_id())
            .cloned()
            .unwrap_or_else(|| PlanView::python_ident(view.contract_id()));

        let mut body = String::new();
        body.push_str(&view.header_comment("Dagster"));
        body.push('\n');
        body.push_str("from dagster import Definitions, OpExecutionContext, graph, op\n\n");

        for step_id in view.step_order() {
            let fn_name = &idents[step_id];
            let step = view.step(step_id);
            let contract = step
                .and_then(|s| s.contract_ref.as_deref().or(s.transform_ref.as_deref()))
                .unwrap_or("-");
            let step_type = step.map(|s| s.step_type.as_str()).unwrap_or("unknown");
            let preds = view.predecessors(step_id);
            body.push_str("@op(name=\"");
            body.push_str(&PlanView::py_string(step_id));
            body.push_str("\", tags={\"dpcs.contractRef\": \"");
            body.push_str(&PlanView::py_string(contract));
            body.push_str("\", \"dpcs.stepType\": \"");
            body.push_str(&PlanView::py_string(step_type));
            body.push_str("\"})\ndef ");
            body.push_str(fn_name);
            if preds.is_empty() {
                body.push_str("_op(context: OpExecutionContext):\n");
            } else {
                body.push_str("_op(context: OpExecutionContext, upstream):\n");
            }
            body.push_str("    context.log.info(\"DPCS step ");
            body.push_str(&PlanView::py_string(step_id));
            body.push_str("\")\n    return True\n\n");
        }

        body.push_str("@graph(name=\"");
        body.push_str(&PlanView::py_string(&job_name));
        body.push_str("_graph\")\ndef ");
        body.push_str(&job_name);
        body.push_str("_graph():\n");

        if view.step_order().is_empty() {
            body.push_str("    pass\n");
        } else if view.dependency_edges().is_empty() {
            // Independent entry ops — do not invent linear dependencies.
            for step_id in view.step_order() {
                let fn_name = &idents[step_id];
                body.push_str("    ");
                body.push_str(fn_name);
                body.push_str("_result = ");
                body.push_str(fn_name);
                body.push_str("_op()\n");
            }
        } else {
            for step_id in view.step_order() {
                let fn_name = &idents[step_id];
                let preds = view.predecessors(step_id);
                body.push_str("    ");
                body.push_str(fn_name);
                body.push_str("_result = ");
                body.push_str(fn_name);
                body.push_str("_op(");
                if preds.is_empty() {
                    body.push(')');
                } else if preds.len() == 1 {
                    body.push_str(&idents[preds[0]]);
                    body.push_str("_result)");
                } else {
                    body.push('[');
                    for (i, pred) in preds.iter().enumerate() {
                        if i > 0 {
                            body.push_str(", ");
                        }
                        body.push_str(&idents[*pred]);
                        body.push_str("_result");
                    }
                    body.push_str("])");
                }
                body.push('\n');
            }
        }

        body.push('\n');
        body.push_str(&job_name);
        body.push_str("_job = ");
        body.push_str(&job_name);
        body.push_str("_graph.to_job(name=\"");
        body.push_str(&PlanView::py_string(&job_name));
        body.push_str("\")\n\ndefs = Definitions(jobs=[");
        body.push_str(&job_name);
        body.push_str("_job])\n");

        Ok(vec![
            python_file("definitions.py", body),
            semantics_file(&view),
        ])
    }
}