scirs2_io/workflow/
functions_2.rs

1//! Auto-generated module
2//!
3//! 🤖 Generated with [SplitRS](https://github.com/cool-japan/splitrs)
4
5use crate::error::{IoError, Result};
6use chrono::{DateTime, Datelike, Duration, Utc};
7use std::collections::{HashMap, HashSet};
8
9use super::types::{Task, Workflow, WorkflowStatus};
10
11/// External workflow engine integration
12pub mod engines {
13    use super::*;
14    /// Trait for external workflow engine adapters
15    pub trait WorkflowEngineAdapter: Send + Sync {
16        /// Convert internal workflow to engine-specific format
17        fn exportworkflow(&self, workflow: &Workflow) -> Result<String>;
18        /// Import workflow from engine-specific format
19        fn importworkflow(&self, definition: &str) -> Result<Workflow>;
20        /// Submit workflow for execution
21        fn submit(&self, workflow: &Workflow) -> Result<String>;
22        /// Get execution status
23        fn get_status(&self, executionid: &str) -> Result<WorkflowStatus>;
24        /// Cancel execution
25        fn cancel(&self, executionid: &str) -> Result<()>;
26    }
27    /// Apache Airflow adapter
28    pub struct AirflowAdapter {
29        api_url: String,
30        auth_token: Option<String>,
31    }
32    impl AirflowAdapter {
33        pub fn new(api_url: impl Into<String>) -> Self {
34            Self {
35                api_url: api_url.into(),
36                auth_token: None,
37            }
38        }
39        pub fn with_auth(mut self, token: impl Into<String>) -> Self {
40            self.auth_token = Some(token.into());
41            self
42        }
43    }
44    impl WorkflowEngineAdapter for AirflowAdapter {
45        fn exportworkflow(&self, workflow: &Workflow) -> Result<String> {
46            let mut dag_code = String::new();
47            dag_code.push_str("from airflow import DAG\n");
48            dag_code.push_str("from airflow.operators.python import PythonOperator\n");
49            dag_code.push_str("from datetime import datetime, timedelta\n\n");
50            dag_code.push_str("dag = DAG(\n");
51            dag_code.push_str(&format!("    '{}',\n", workflow.id));
52            dag_code.push_str(&format!(
53                "    description='{}',\n",
54                workflow.description.as_deref().unwrap_or("")
55            ));
56            dag_code.push_str("    default_args={\n");
57            dag_code.push_str("        'owner': 'scirs2',\n");
58            dag_code.push_str("        'retries': 3,\n");
59            dag_code.push_str("        'retry_delay': timedelta(minutes=5),\n");
60            dag_code.push_str("    },\n");
61            dag_code.push_str("    schedule_interval=None,\n");
62            dag_code.push_str("    start_date=datetime(2024, 1, 1),\n");
63            dag_code.push_str("    catchup=False,\n");
64            dag_code.push_str(")\n\n");
65            for task in &workflow.tasks {
66                dag_code.push_str(&format!("{} = PythonOperator(\n", task.id));
67                dag_code.push_str(&format!("    task_id='{}',\n", task.id));
68                dag_code.push_str(&format!(
69                    "    python_callable=lambda: print('{}'),\n",
70                    task.name
71                ));
72                dag_code.push_str("    dag=dag,\n");
73                dag_code.push_str(")\n\n");
74            }
75            for (task_id, deps) in &workflow.dependencies {
76                for dep in deps {
77                    dag_code.push_str(&format!("{dep} >> {task_id}\n"));
78                }
79            }
80            Ok(dag_code)
81        }
82        fn importworkflow(&self, definition: &str) -> Result<Workflow> {
83            // Round-trip parser for the Airflow DAG format produced by exportworkflow.
84            // Extracts: DAG id, task ids + names, and dependency edges (dep >> task_id).
85            use super::super::types::{ResourceRequirements, TaskType, WorkflowConfig};
86            use crate::metadata::Metadata;
87
88            let mut workflow_id = String::from("imported_dag");
89            let mut tasks = Vec::new();
90            let mut dependencies: HashMap<String, Vec<String>> = HashMap::new();
91            // State tracking for multi-line DAG(...) constructor
92            let mut inside_dag_constructor = false;
93            let mut dag_id_found = false;
94
95            for line in definition.lines() {
96                let trimmed = line.trim();
97
98                // The exporter emits:
99                //   dag = DAG(
100                //       'dag_id',
101                // Handle both single-line `dag = DAG('id', ...)` and multi-line forms.
102                if trimmed == "dag = DAG(" {
103                    inside_dag_constructor = true;
104                    continue;
105                }
106                // Single-line form: dag = DAG('dag_id', ...)
107                if trimmed.starts_with("dag = DAG('") {
108                    if let Some(inner) = trimmed.strip_prefix("dag = DAG('") {
109                        if let Some(end) = inner.find('\'') {
110                            workflow_id = inner[..end].to_string();
111                            dag_id_found = true;
112                        }
113                    }
114                }
115                // Inside multi-line DAG constructor: first quoted string is the dag id
116                if inside_dag_constructor && !dag_id_found {
117                    if trimmed.starts_with('\'') {
118                        let inner = trimmed.trim_start_matches('\'');
119                        if let Some(end) = inner.find('\'') {
120                            workflow_id = inner[..end].to_string();
121                            dag_id_found = true;
122                        }
123                    }
124                    // Detect end of constructor
125                    if trimmed == ")" {
126                        inside_dag_constructor = false;
127                    }
128                }
129
130                // Extract task: task_id = PythonOperator(
131                // followed by:  task_id='task_id', then python_callable=lambda: print('task_name')
132                if trimmed.contains(" = PythonOperator(") {
133                    if let Some(eq_pos) = trimmed.find(" = PythonOperator(") {
134                        let task_id = trimmed[..eq_pos].trim().to_string();
135                        if !task_id.is_empty() {
136                            tasks.push((task_id, String::new()));
137                        }
138                    }
139                }
140
141                // Extract task name from: python_callable=lambda: print('task_name'),
142                if trimmed.starts_with("python_callable=lambda: print('") {
143                    if let Some(stripped) = trimmed.strip_prefix("python_callable=lambda: print('")
144                    {
145                        if let Some(end) = stripped.find('\'') {
146                            let task_name = stripped[..end].to_string();
147                            if let Some(last) = tasks.last_mut() {
148                                last.1 = task_name;
149                            }
150                        }
151                    }
152                }
153
154                // Extract dependency: dep >> task_id
155                if trimmed.contains(" >> ") && !trimmed.starts_with('#') {
156                    let parts: Vec<&str> = trimmed.split(" >> ").collect();
157                    if parts.len() == 2 {
158                        let dep = parts[0].trim().to_string();
159                        let task_id = parts[1].trim().to_string();
160                        if !dep.is_empty() && !task_id.is_empty() {
161                            dependencies.entry(task_id).or_default().push(dep);
162                        }
163                    }
164                }
165            }
166
167            let task_objects: Vec<Task> = tasks
168                .into_iter()
169                .map(|(id, name)| {
170                    let display_name = if name.is_empty() { id.clone() } else { name };
171                    Task {
172                        id,
173                        name: display_name,
174                        task_type: TaskType::Script,
175                        config: serde_json::Value::Null,
176                        inputs: Vec::new(),
177                        outputs: Vec::new(),
178                        resources: ResourceRequirements::default(),
179                    }
180                })
181                .collect();
182
183            Ok(Workflow {
184                id: workflow_id.clone(),
185                name: workflow_id,
186                description: None,
187                tasks: task_objects,
188                dependencies,
189                config: WorkflowConfig::default(),
190                metadata: Metadata::new(),
191            })
192        }
193        fn submit(&self, workflow: &Workflow) -> Result<String> {
194            let executionid = format!("{}_run_{}", workflow.id, Utc::now().timestamp());
195            Ok(executionid)
196        }
197        fn get_status(&self, _executionid: &str) -> Result<WorkflowStatus> {
198            Ok(WorkflowStatus::Running)
199        }
200        fn cancel(&self, _executionid: &str) -> Result<()> {
201            Ok(())
202        }
203    }
204    /// Prefect adapter
205    pub struct PrefectAdapter {
206        api_url: String,
207        project_name: String,
208    }
209    impl PrefectAdapter {
210        pub fn new(api_url: impl Into<String>, project: impl Into<String>) -> Self {
211            Self {
212                api_url: api_url.into(),
213                project_name: project.into(),
214            }
215        }
216    }
217    impl WorkflowEngineAdapter for PrefectAdapter {
218        fn exportworkflow(&self, workflow: &Workflow) -> Result<String> {
219            let mut flow_code = String::new();
220            flow_code.push_str("from prefect import flow, task\n");
221            flow_code.push_str("from prefect.task_runners import SequentialTaskRunner\n\n");
222            for task in &workflow.tasks {
223                flow_code.push_str(&format!("@task(name='{}')\n", task.name));
224                flow_code.push_str(&format!("def {}():\n", task.id));
225                flow_code.push_str(&format!("    print('Executing {}')\n", task.name));
226                flow_code.push_str("    return True\n\n");
227            }
228            flow_code.push_str(&format!(
229                "@flow(name='{}', task_runner=SequentialTaskRunner())\n",
230                workflow.name
231            ));
232            flow_code.push_str("def workflow_flow():\n");
233            let mut executed = HashSet::new();
234            let mut to_execute: Vec<_> = workflow.tasks.iter().map(|t| &t.id).collect();
235            while !to_execute.is_empty() {
236                let mut progress = false;
237                to_execute.retain(|task_id| {
238                    let deps = workflow.dependencies.get(*task_id);
239                    let can_execute =
240                        deps.is_none_or(|d| d.iter().all(|dep| executed.contains(dep)));
241                    if can_execute {
242                        flow_code.push_str(&format!("    {task_id}()\n"));
243                        executed.insert((*task_id).clone());
244                        progress = true;
245                        false
246                    } else {
247                        true
248                    }
249                });
250                if !progress && !to_execute.is_empty() {
251                    return Err(IoError::Other("Circular dependency detected".to_string()));
252                }
253            }
254            flow_code.push_str("\nif __name__ == '__main__':\n");
255            flow_code.push_str("    workflow_flow()\n");
256            Ok(flow_code)
257        }
258        fn importworkflow(&self, definition: &str) -> Result<Workflow> {
259            // Round-trip parser for the Prefect flow format produced by exportworkflow.
260            // Extracts: flow name, task ids + names, and call order from workflow_flow().
261            use super::super::types::{ResourceRequirements, TaskType, WorkflowConfig};
262            use crate::metadata::Metadata;
263
264            let mut workflow_name = String::from("imported_flow");
265            let mut workflow_id = String::from("imported_flow");
266            // Map from task function id -> task name (from @task(name='...'))
267            let mut pending_task_name: Option<String> = None;
268            let mut task_name_map: HashMap<String, String> = HashMap::new();
269            // Task call order extracted from workflow_flow()
270            let mut ordered_task_ids: Vec<String> = Vec::new();
271            let mut inside_flow_body = false;
272
273            for line in definition.lines() {
274                let trimmed = line.trim();
275
276                // @flow(name='workflow_name', ...)
277                if trimmed.starts_with("@flow(name='") {
278                    if let Some(stripped) = trimmed.strip_prefix("@flow(name='") {
279                        if let Some(end) = stripped.find('\'') {
280                            workflow_name = stripped[..end].to_string();
281                            workflow_id = workflow_name.replace(' ', "_").to_lowercase();
282                        }
283                    }
284                }
285
286                // @task(name='task_name')
287                if trimmed.starts_with("@task(name='") {
288                    if let Some(stripped) = trimmed.strip_prefix("@task(name='") {
289                        if let Some(end) = stripped.find('\'') {
290                            pending_task_name = Some(stripped[..end].to_string());
291                        }
292                    }
293                }
294
295                // def task_id():  — task function definition
296                if trimmed.starts_with("def ")
297                    && trimmed.ends_with("():")
298                    && !trimmed.starts_with("def workflow_flow")
299                {
300                    if let Some(stripped) = trimmed.strip_prefix("def ") {
301                        if let Some(end) = stripped.find("():") {
302                            let task_id = stripped[..end].to_string();
303                            let name = pending_task_name.take().unwrap_or_else(|| task_id.clone());
304                            task_name_map.insert(task_id, name);
305                        }
306                    }
307                }
308
309                // def workflow_flow(): — start of flow body
310                if trimmed == "def workflow_flow():" {
311                    inside_flow_body = true;
312                    continue;
313                }
314
315                // Inside flow body: lines like "    task_id()"
316                if inside_flow_body {
317                    if trimmed.starts_with("if __name__") {
318                        inside_flow_body = false;
319                    } else if trimmed.ends_with("()") {
320                        let task_id = trimmed.trim_end_matches("()").to_string();
321                        if !task_id.is_empty() && task_name_map.contains_key(&task_id) {
322                            ordered_task_ids.push(task_id);
323                        }
324                    }
325                }
326            }
327
328            let tasks: Vec<Task> = ordered_task_ids
329                .iter()
330                .map(|id| {
331                    let name = task_name_map.get(id).cloned().unwrap_or_else(|| id.clone());
332                    Task {
333                        id: id.clone(),
334                        name,
335                        task_type: TaskType::Script,
336                        config: serde_json::Value::Null,
337                        inputs: Vec::new(),
338                        outputs: Vec::new(),
339                        resources: ResourceRequirements::default(),
340                    }
341                })
342                .collect();
343
344            // Prefect exporter serialises tasks in topological order inline (no explicit deps).
345            // The ordering itself encodes sequencing but we cannot recover edge data without
346            // more information — return empty dependency map as the safe round-trip value.
347            Ok(Workflow {
348                id: workflow_id,
349                name: workflow_name,
350                description: None,
351                tasks,
352                dependencies: HashMap::new(),
353                config: WorkflowConfig::default(),
354                metadata: Metadata::new(),
355            })
356        }
357        fn submit(&self, workflow: &Workflow) -> Result<String> {
358            let flow_run_id = uuid::Uuid::new_v4().to_string();
359            Ok(flow_run_id)
360        }
361        fn get_status(&self, _executionid: &str) -> Result<WorkflowStatus> {
362            Ok(WorkflowStatus::Running)
363        }
364        fn cancel(&self, _executionid: &str) -> Result<()> {
365            Ok(())
366        }
367    }
368    /// Dagster adapter
369    pub struct DagsterAdapter {
370        repository_url: String,
371    }
372    impl DagsterAdapter {
373        /// Create a new Dagster adapter
374        pub fn new(repository_url: impl Into<String>) -> Self {
375            Self {
376                repository_url: repository_url.into(),
377            }
378        }
379    }
380    impl WorkflowEngineAdapter for DagsterAdapter {
381        fn exportworkflow(&self, workflow: &Workflow) -> Result<String> {
382            let mut job_code = String::new();
383            job_code.push_str("from dagster import job, op, Config\n\n");
384            for task in &workflow.tasks {
385                job_code.push_str(&format!("@op(name='{}')\n", task.id));
386                job_code.push_str(&format!("def {}(context):\n", task.id));
387                job_code.push_str(&format!(
388                    "    context.log.info('Executing {}')\n",
389                    task.name
390                ));
391                job_code.push_str("    return True\n\n");
392            }
393            job_code.push_str(&format!("@job(name='{}')\n", workflow.id));
394            job_code.push_str("def workflow_job():\n");
395            for task in &workflow.tasks {
396                if let Some(deps) = workflow.dependencies.get(&task.id) {
397                    let deps_str = deps.join(", ");
398                    job_code.push_str(&format!("    {}({}())\n", task.id, deps_str));
399                } else {
400                    job_code.push_str(&format!("    {}()\n", task.id));
401                }
402            }
403            Ok(job_code)
404        }
405        fn importworkflow(&self, definition: &str) -> Result<Workflow> {
406            // Round-trip parser for the Dagster job format produced by exportworkflow.
407            // Extracts: job name (workflow id), op names, and call-graph deps from workflow_job().
408            use super::super::types::{ResourceRequirements, TaskType, WorkflowConfig};
409            use crate::metadata::Metadata;
410
411            let mut workflow_id = String::from("imported_job");
412            let mut pending_op_id: Option<String> = None;
413            // Map task_id -> task_name from @op(name='...') + context.log.info line
414            let mut task_info: Vec<(String, String)> = Vec::new();
415            let mut dependencies: HashMap<String, Vec<String>> = HashMap::new();
416            let mut inside_job_body = false;
417
418            for line in definition.lines() {
419                let trimmed = line.trim();
420
421                // @job(name='job_id')
422                if trimmed.starts_with("@job(name='") {
423                    if let Some(stripped) = trimmed.strip_prefix("@job(name='") {
424                        if let Some(end) = stripped.find('\'') {
425                            workflow_id = stripped[..end].to_string();
426                        }
427                    }
428                }
429
430                // @op(name='task_id')
431                if trimmed.starts_with("@op(name='") {
432                    if let Some(stripped) = trimmed.strip_prefix("@op(name='") {
433                        if let Some(end) = stripped.find('\'') {
434                            pending_op_id = Some(stripped[..end].to_string());
435                        }
436                    }
437                }
438
439                // context.log.info('Executing task_name')
440                if trimmed.starts_with("context.log.info('Executing ") {
441                    if let Some(stripped) = trimmed.strip_prefix("context.log.info('Executing ") {
442                        if let Some(end) = stripped.find('\'') {
443                            let task_name = stripped[..end].to_string();
444                            if let Some(op_id) = pending_op_id.take() {
445                                task_info.push((op_id, task_name));
446                            }
447                        }
448                    }
449                }
450
451                // def workflow_job():
452                if trimmed == "def workflow_job():" {
453                    inside_job_body = true;
454                    continue;
455                }
456
457                // Inside job body: lines like "    task_id(dep1(), dep2())" or "    task_id()"
458                if inside_job_body && trimmed.starts_with(|c: char| c.is_alphabetic() || c == '_') {
459                    // Parse: task_id(dep_calls) where dep_calls may be empty or "dep1(), dep2()"
460                    if let Some(paren_pos) = trimmed.find('(') {
461                        let task_id = trimmed[..paren_pos].trim().to_string();
462                        let rest = &trimmed[paren_pos + 1..];
463                        // Remove trailing ')'
464                        let args = rest.trim_end_matches(')').trim();
465                        if !args.is_empty() {
466                            // Each dep looks like "dep_id()" — extract dep_id
467                            let deps: Vec<String> = args
468                                .split(',')
469                                .filter_map(|seg| {
470                                    let seg = seg.trim();
471                                    // dep calls look like "dep_id()"
472                                    seg.find('(').map(|p| seg[..p].trim().to_string())
473                                })
474                                .filter(|s| !s.is_empty())
475                                .collect();
476                            if !deps.is_empty() {
477                                dependencies.insert(task_id, deps);
478                            }
479                        }
480                    }
481                }
482            }
483
484            let tasks: Vec<Task> = task_info
485                .into_iter()
486                .map(|(id, name)| Task {
487                    id,
488                    name,
489                    task_type: TaskType::Script,
490                    config: serde_json::Value::Null,
491                    inputs: Vec::new(),
492                    outputs: Vec::new(),
493                    resources: ResourceRequirements::default(),
494                })
495                .collect();
496
497            Ok(Workflow {
498                id: workflow_id.clone(),
499                name: workflow_id,
500                description: None,
501                tasks,
502                dependencies,
503                config: WorkflowConfig::default(),
504                metadata: Metadata::new(),
505            })
506        }
507        fn submit(&self, workflow: &Workflow) -> Result<String> {
508            Ok(uuid::Uuid::new_v4().to_string())
509        }
510        fn get_status(&self, _executionid: &str) -> Result<WorkflowStatus> {
511            Ok(WorkflowStatus::Running)
512        }
513        fn cancel(&self, _executionid: &str) -> Result<()> {
514            Ok(())
515        }
516    }
517}
518/// Dynamic workflow generation
519pub mod dynamic {
520    use super::*;
521    /// Dynamic workflow generator
522    pub struct DynamicWorkflowGenerator {
523        templates: HashMap<String, WorkflowTemplate>,
524    }
525    #[derive(Debug, Clone)]
526    pub struct WorkflowTemplate {
527        pub baseworkflow: Workflow,
528        pub parameters: Vec<ParameterDef>,
529        pub generators: Vec<TaskGenerator>,
530    }
531    #[derive(Debug, Clone)]
532    pub struct ParameterDef {
533        pub name: String,
534        pub param_type: ParameterType,
535        pub required: bool,
536        pub default: Option<serde_json::Value>,
537    }
538    #[derive(Debug, Clone)]
539    pub enum ParameterType {
540        String,
541        Integer,
542        Float,
543        Boolean,
544        List(Box<ParameterType>),
545        Object,
546    }
547    #[derive(Debug, Clone)]
548    pub enum TaskGenerator {
549        ForEach {
550            parameter: String,
551            task_template: Task,
552        },
553        Conditional {
554            condition: String,
555            true_tasks: Vec<Task>,
556            false_tasks: Vec<Task>,
557        },
558        Repeat {
559            count_param: String,
560            task_template: Task,
561        },
562    }
563    impl Default for DynamicWorkflowGenerator {
564        fn default() -> Self {
565            Self::new()
566        }
567    }
568    impl DynamicWorkflowGenerator {
569        pub fn new() -> Self {
570            Self {
571                templates: HashMap::new(),
572            }
573        }
574        /// Register a workflow template
575        pub fn register_template(&mut self, name: impl Into<String>, template: WorkflowTemplate) {
576            self.templates.insert(name.into(), template);
577        }
578        /// Generate workflow from template
579        pub fn generate(
580            &self,
581            template_name: &str,
582            params: HashMap<String, serde_json::Value>,
583        ) -> Result<Workflow> {
584            let template = self.templates.get(template_name).ok_or_else(|| {
585                IoError::NotFound(format!("Template '{template_name}' not found"))
586            })?;
587            for param_def in &template.parameters {
588                if param_def.required && !params.contains_key(&param_def.name) {
589                    return Err(IoError::ValidationError(format!(
590                        "Required parameter '{}' not provided",
591                        param_def.name
592                    )));
593                }
594            }
595            let mut workflow = template.baseworkflow.clone();
596            workflow.id = format!("{}_{}", workflow.id, Utc::now().timestamp());
597            for generator in &template.generators {
598                self.apply_generator(&mut workflow, generator, &params)?;
599            }
600            Ok(workflow)
601        }
602        fn apply_generator(
603            &self,
604            workflow: &mut Workflow,
605            generator: &TaskGenerator,
606            params: &HashMap<String, serde_json::Value>,
607        ) -> Result<()> {
608            match generator {
609                TaskGenerator::ForEach {
610                    parameter,
611                    task_template,
612                } => {
613                    if let Some(serde_json::Value::Array(items)) = params.get(parameter) {
614                        for (i, item) in items.iter().enumerate() {
615                            let mut task = task_template.clone();
616                            task.id = format!("{}_{}", task.id, i);
617                            task.name = format!("{} [{}]", task.name, i);
618                            if let serde_json::Value::Object(mut config) = task.config.clone() {
619                                config.insert("item".to_string(), item.clone());
620                                task.config = serde_json::Value::Object(config);
621                            }
622                            workflow.tasks.push(task);
623                        }
624                    }
625                }
626                TaskGenerator::Conditional {
627                    condition,
628                    true_tasks,
629                    false_tasks,
630                } => {
631                    let condition_result = self.evaluate_condition(condition, params)?;
632                    if condition_result {
633                        workflow.tasks.extend(true_tasks.iter().cloned());
634                    } else {
635                        workflow.tasks.extend(false_tasks.iter().cloned());
636                    }
637                }
638                TaskGenerator::Repeat {
639                    count_param,
640                    task_template,
641                } => {
642                    if let Some(serde_json::Value::Number(n)) = params.get(count_param) {
643                        if let Some(count) = n.as_u64() {
644                            for i in 0..count {
645                                let mut task = task_template.clone();
646                                task.id = format!("{}_{}", task.id, i);
647                                task.name = format!("{} [{}]", task.name, i);
648                                workflow.tasks.push(task);
649                            }
650                        }
651                    }
652                }
653            }
654            Ok(())
655        }
656        fn evaluate_condition(
657            &self,
658            condition: &str,
659            params: &HashMap<String, serde_json::Value>,
660        ) -> Result<bool> {
661            if let Some((param, value)) = condition.split_once("==") {
662                let param = param.trim();
663                let value = value.trim().trim_matches('"');
664                if let Some(serde_json::Value::String(s)) = params.get(param) {
665                    return Ok(s == value);
666                }
667            }
668            Ok(false)
669        }
670    }
671}
672
673#[cfg(test)]
674mod tests {
675    use super::super::types::{ResourceRequirements, Task, TaskType, Workflow, WorkflowConfig};
676    use super::*;
677    use crate::metadata::Metadata;
678
679    fn make_workflow(id: &str, tasks: Vec<Task>) -> Workflow {
680        Workflow {
681            id: id.to_string(),
682            name: id.to_string(),
683            description: None,
684            tasks,
685            dependencies: HashMap::new(),
686            config: WorkflowConfig::default(),
687            metadata: Metadata::new(),
688        }
689    }
690
691    fn make_task(id: &str, name: &str) -> Task {
692        Task {
693            id: id.to_string(),
694            name: name.to_string(),
695            task_type: TaskType::Script,
696            config: serde_json::Value::Null,
697            inputs: Vec::new(),
698            outputs: Vec::new(),
699            resources: ResourceRequirements::default(),
700        }
701    }
702
703    #[test]
704    fn test_airflow_import_roundtrip() {
705        use engines::{AirflowAdapter, WorkflowEngineAdapter};
706        let adapter = AirflowAdapter::new("http://localhost:8080");
707        let tasks = vec![
708            make_task("ingest", "Ingest Data"),
709            make_task("transform", "Transform Data"),
710        ];
711        let workflow = make_workflow("my_pipeline", tasks);
712        let exported = adapter
713            .exportworkflow(&workflow)
714            .expect("Export should succeed");
715        let imported = adapter
716            .importworkflow(&exported)
717            .expect("Import should succeed");
718        assert_eq!(imported.id, "my_pipeline");
719        assert_eq!(imported.tasks.len(), 2);
720        let ids: Vec<&str> = imported.tasks.iter().map(|t| t.id.as_str()).collect();
721        assert!(ids.contains(&"ingest"), "Expected 'ingest' in {:?}", ids);
722        assert!(
723            ids.contains(&"transform"),
724            "Expected 'transform' in {:?}",
725            ids
726        );
727    }
728
729    #[test]
730    fn test_prefect_import_roundtrip() {
731        use engines::{PrefectAdapter, WorkflowEngineAdapter};
732        let adapter = PrefectAdapter::new("http://localhost:4200", "test_project");
733        let tasks = vec![
734            make_task("fetch", "Fetch Records"),
735            make_task("validate", "Validate Records"),
736        ];
737        let workflow = make_workflow("data_flow", tasks);
738        let exported = adapter
739            .exportworkflow(&workflow)
740            .expect("Export should succeed");
741        let imported = adapter
742            .importworkflow(&exported)
743            .expect("Import should succeed");
744        assert!(imported.tasks.len() >= 1, "Should have at least one task");
745    }
746
747    #[test]
748    fn test_dagster_import_roundtrip() {
749        use engines::{DagsterAdapter, WorkflowEngineAdapter};
750        let adapter = DagsterAdapter::new("");
751        let tasks = vec![
752            make_task("load", "Load Source"),
753            make_task("process", "Process Data"),
754        ];
755        let mut workflow = make_workflow("etl_job", tasks);
756        workflow
757            .dependencies
758            .insert("process".to_string(), vec!["load".to_string()]);
759        let exported = adapter
760            .exportworkflow(&workflow)
761            .expect("Export should succeed");
762        let imported = adapter
763            .importworkflow(&exported)
764            .expect("Import should succeed");
765        assert_eq!(imported.id, "etl_job");
766        assert!(imported.tasks.len() >= 2, "Should have at least 2 tasks");
767        // Dependency: process depends on load
768        if let Some(deps) = imported.dependencies.get("process") {
769            assert!(deps.contains(&"load".to_string()));
770        }
771    }
772}
scirs2_io/workflow/functions_2.rs

scirs2_io/workflow/
functions_2.rs
