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

//! Temporal workflow stub adapter (experimental).

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};

/// Temporal binding adapter (experimental).
pub struct TemporalAdapter;

impl OrchestratorAdapter for TemporalAdapter {
    fn target(&self) -> BindingTarget {
        BindingTarget::Temporal
    }

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

        let mut body = String::new();
        body.push_str(&view.header_comment("Temporal (experimental)"));
        body.push('\n');
        body.push_str(
            "\
from datetime import timedelta

from temporalio import activity, workflow

",
        );

        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("-");
            body.push_str("@activity.defn(name=\"");
            body.push_str(&PlanView::py_string(step_id));
            body.push_str("\")\nasync def ");
            body.push_str(fn_name);
            body.push_str("_activity() -> None:\n    \"\"\"DPCS step ");
            body.push_str(&PlanView::py_string(step_id));
            body.push_str(" contractRef=");
            body.push_str(&PlanView::py_string(contract));
            body.push_str(".\"\"\"\n    return None\n\n");
        }

        body.push_str("@workflow.defn(name=\"");
        body.push_str(&PlanView::py_string(view.contract_id()));
        body.push_str("\")\nclass ");
        body.push_str(&class_name);
        body.push_str(
            ":
    @workflow.run
    async def run(self) -> None:
",
        );

        if view.step_order().is_empty() {
            body.push_str("        return None\n");
        } else if view.dependency_edges().is_empty() {
            body.push_str(
                "        # Independent activities: sequential awaits are a linearization only.\n",
            );
            for step_id in view.step_order() {
                let fn_name = &idents[step_id];
                body.push_str("        await workflow.execute_activity(\n            ");
                body.push_str(fn_name);
                body.push_str(
                    "_activity,\n            start_to_close_timeout=timedelta(minutes=30),\n        )\n",
                );
            }
        } else {
            body.push_str(
                "        # Await activities in topologic step_order; edges are enforced by sequencing.\n",
            );
            for step_id in view.step_order() {
                let fn_name = &idents[step_id];
                let preds = view.predecessors(step_id);
                if !preds.is_empty() {
                    body.push_str("        # waits for: ");
                    body.push_str(&preds.join(", "));
                    body.push('\n');
                }
                body.push_str("        await workflow.execute_activity(\n            ");
                body.push_str(fn_name);
                body.push_str(
                    "_activity,\n            start_to_close_timeout=timedelta(minutes=30),\n        )\n",
                );
            }
        }

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