scirs2_io/workflow/
functions.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 serde::{Deserialize, Serialize};
8use serde_json::json;
9use std::collections::{HashMap, HashSet};
10use std::sync::{Arc, Mutex};
11
12use super::types::{
13    ResourceRequirements, ScheduleConfig, Task, TaskType, Workflow, WorkflowBuilder,
14    WorkflowExecutor, WorkflowState, WorkflowStatus,
15};
16
17/// Task builders for common operations
18pub mod tasks {
19    use super::*;
20    /// Create a data ingestion task
21    pub fn data_ingestion(id: impl Into<String>, name: impl Into<String>) -> TaskBuilder {
22        TaskBuilder::new(id, name, TaskType::DataIngestion)
23    }
24    /// Create a transformation task
25    pub fn transform(id: impl Into<String>, name: impl Into<String>) -> TaskBuilder {
26        TaskBuilder::new(id, name, TaskType::Transform)
27    }
28    /// Create a validation task
29    pub fn validation(id: impl Into<String>, name: impl Into<String>) -> TaskBuilder {
30        TaskBuilder::new(id, name, TaskType::Validation)
31    }
32    /// Create an export task
33    pub fn export(id: impl Into<String>, name: impl Into<String>) -> TaskBuilder {
34        TaskBuilder::new(id, name, TaskType::Export)
35    }
36    /// Task builder
37    pub struct TaskBuilder {
38        task: Task,
39    }
40    impl TaskBuilder {
41        pub fn new(id: impl Into<String>, name: impl Into<String>, task_type: TaskType) -> Self {
42            Self {
43                task: Task {
44                    id: id.into(),
45                    name: name.into(),
46                    task_type,
47                    config: json!({}),
48                    inputs: Vec::new(),
49                    outputs: Vec::new(),
50                    resources: ResourceRequirements::default(),
51                },
52            }
53        }
54        pub fn config(mut self, config: serde_json::Value) -> Self {
55            self.task.config = config;
56            self
57        }
58        pub fn input(mut self, input: impl Into<String>) -> Self {
59            self.task.inputs.push(input.into());
60            self
61        }
62        pub fn output(mut self, output: impl Into<String>) -> Self {
63            self.task.outputs.push(output.into());
64            self
65        }
66        pub fn resources(mut self, cpu: usize, memorygb: f64) -> Self {
67            self.task.resources.cpu_cores = Some(cpu);
68            self.task.resources.memorygb = Some(memorygb);
69            self
70        }
71        pub fn build(self) -> Task {
72            self.task
73        }
74    }
75}
76/// Workflow templates for common patterns
77pub mod templates {
78    use super::*;
79    /// Create an ETL (Extract-Transform-Load) workflow
80    pub fn etlworkflow(name: impl Into<String>) -> WorkflowBuilder {
81        let _name = name.into();
82        let id = format!("etl_{}", Utc::now().timestamp());
83        WorkflowBuilder::new(&id, &_name)
84            .description("Standard ETL workflow template")
85            .add_task(
86                tasks::data_ingestion("extract", "Extract Data")
87                    .config(serde_json::json!(
88                        { "source" : "database", "query" : "SELECT * FROM raw_data" }
89                    ))
90                    .output("raw_data")
91                    .build(),
92            )
93            .add_task(
94                tasks::transform("transform", "Transform Data")
95                    .input("raw_data")
96                    .output("transformed_data")
97                    .config(serde_json::json!(
98                        { "operations" : ["normalize", "aggregate", "filter"] }
99                    ))
100                    .build(),
101            )
102            .add_task(
103                tasks::validation("validate", "Validate Data")
104                    .input("transformed_data")
105                    .output("validated_data")
106                    .build(),
107            )
108            .add_task(
109                tasks::export("load", "Load Data")
110                    .input("validated_data")
111                    .config(serde_json::json!(
112                        { "destination" : "warehouse", "table" : "processed_data" }
113                    ))
114                    .build(),
115            )
116            .add_dependency("transform", "extract")
117            .add_dependency("validate", "transform")
118            .add_dependency("load", "validate")
119    }
120    /// Create a batch processing workflow
121    pub fn batch_processing(name: impl Into<String>, _batch_size: usize) -> WorkflowBuilder {
122        let name = name.into();
123        let id = format!("batch_{}", Utc::now().timestamp());
124        WorkflowBuilder::new(&id, &name)
125            .description("Batch processing workflow template")
126            .configure(|config| {
127                config.max_parallel_tasks = 8;
128                config.scheduling = Some(ScheduleConfig {
129                    cron: Some("0 2 * * *".to_string()),
130                    interval: None,
131                    start_time: None,
132                    end_time: None,
133                });
134            })
135    }
136}
137/// Workflow monitoring and metrics
138pub mod monitoring {
139    use super::*;
140    #[derive(Debug, Clone, Serialize, Deserialize)]
141    pub struct WorkflowMetrics {
142        pub total_executions: usize,
143        pub successful_executions: usize,
144        pub failed_executions: usize,
145        pub average_duration: Duration,
146        pub task_metrics: HashMap<String, TaskMetrics>,
147    }
148    #[derive(Debug, Clone, Serialize, Deserialize)]
149    pub struct TaskMetrics {
150        pub total_runs: usize,
151        pub success_rate: f64,
152        pub average_duration: Duration,
153        pub retry_rate: f64,
154    }
155    /// Collect metrics for a workflow
156    pub fn collect_metrics(states: &[WorkflowState]) -> WorkflowMetrics {
157        let total = states.len();
158        let successful = states
159            .iter()
160            .filter(|s| s.status == WorkflowStatus::Success)
161            .count();
162        let failed = states
163            .iter()
164            .filter(|s| s.status == WorkflowStatus::Failed)
165            .count();
166        let durations: Vec<Duration> = states
167            .iter()
168            .filter_map(|s| match (s.start_time, s.end_time) {
169                (Some(start), Some(end)) => Some(end - start),
170                _ => None,
171            })
172            .collect();
173        let avg_duration = if durations.is_empty() {
174            Duration::seconds(0)
175        } else {
176            let total_secs: i64 = durations.iter().map(|d| d.num_seconds()).sum();
177            Duration::seconds(total_secs / durations.len() as i64)
178        };
179        WorkflowMetrics {
180            total_executions: total,
181            successful_executions: successful,
182            failed_executions: failed,
183            average_duration: avg_duration,
184            task_metrics: HashMap::new(),
185        }
186    }
187}
188/// Advanced scheduling capabilities
189pub mod scheduling {
190    use super::*;
191    use cron::Schedule as CronSchedule;
192    use std::str::FromStr;
193    /// Advanced scheduler with support for complex scheduling patterns
194    pub struct WorkflowScheduler {
195        schedules: HashMap<String, ScheduledWorkflow>,
196        executor: Arc<WorkflowExecutor>,
197        running: Arc<Mutex<bool>>,
198    }
199    #[derive(Debug)]
200    struct ScheduledWorkflow {
201        workflow: Workflow,
202        schedule: WorkflowSchedule,
203        last_run: Option<DateTime<Utc>>,
204        next_run: Option<DateTime<Utc>>,
205    }
206    #[derive(Debug, Clone)]
207    pub enum WorkflowSchedule {
208        Cron(String),
209        Interval { seconds: u64 },
210        FixedDelay { seconds: u64 },
211        OneTime(DateTime<Utc>),
212        Complex(ComplexSchedule),
213    }
214    #[derive(Debug, Clone)]
215    pub struct ComplexSchedule {
216        pub business_days_only: bool,
217        pub exclude_holidays: bool,
218        pub timezone: String,
219        pub blackout_periods: Vec<(DateTime<Utc>, DateTime<Utc>)>,
220        pub dependencies: Vec<ScheduleDependency>,
221    }
222    #[derive(Debug, Clone)]
223    pub enum ScheduleDependency {
224        FileArrival { path: String, pattern: String },
225        DataAvailability { source: String, threshold: f64 },
226        ExternalTrigger { webhook: String },
227        WorkflowCompletion { workflowid: String },
228    }
229    impl WorkflowScheduler {
230        pub fn new(executor: Arc<WorkflowExecutor>) -> Self {
231            Self {
232                schedules: HashMap::new(),
233                executor,
234                running: Arc::new(Mutex::new(false)),
235            }
236        }
237        /// Schedule a workflow
238        pub fn schedule(&mut self, workflow: Workflow, schedule: WorkflowSchedule) -> Result<()> {
239            let next_run = self.calculate_next_run(&schedule, None)?;
240            self.schedules.insert(
241                workflow.id.clone(),
242                ScheduledWorkflow {
243                    workflow,
244                    schedule,
245                    last_run: None,
246                    next_run,
247                },
248            );
249            Ok(())
250        }
251        /// Calculate next run time based on schedule
252        fn calculate_next_run(
253            &self,
254            schedule: &WorkflowSchedule,
255            last_run: Option<DateTime<Utc>>,
256        ) -> Result<Option<DateTime<Utc>>> {
257            match schedule {
258                WorkflowSchedule::Cron(cron_expr) => {
259                    let schedule = CronSchedule::from_str(cron_expr)
260                        .map_err(|e| IoError::Other(format!("Invalid cron expression: {e}")))?;
261                    let after = last_run.unwrap_or_else(Utc::now);
262                    Ok(schedule.after(&after).next())
263                }
264                WorkflowSchedule::Interval { seconds } => {
265                    let base = last_run.unwrap_or_else(Utc::now);
266                    Ok(Some(base + Duration::seconds(*seconds as i64)))
267                }
268                WorkflowSchedule::FixedDelay { seconds } => {
269                    Ok(Some(Utc::now() + Duration::seconds(*seconds as i64)))
270                }
271                WorkflowSchedule::OneTime(time) => {
272                    if *time > Utc::now() {
273                        Ok(Some(*time))
274                    } else {
275                        Ok(None)
276                    }
277                }
278                WorkflowSchedule::Complex(complex) => {
279                    self.calculate_complex_schedule(complex, last_run)
280                }
281            }
282        }
283        fn calculate_complex_schedule(
284            &self,
285            complex: &ComplexSchedule,
286            last_run: Option<DateTime<Utc>>,
287        ) -> Result<Option<DateTime<Utc>>> {
288            let mut next_run = last_run.unwrap_or_else(Utc::now) + Duration::days(1);
289            if complex.business_days_only {
290                while next_run.weekday().num_days_from_monday() >= 5 {
291                    next_run += Duration::days(1);
292                }
293            }
294            for (start, end) in &complex.blackout_periods {
295                if next_run >= *start && next_run <= *end {
296                    next_run = *end + Duration::seconds(1);
297                }
298            }
299            Ok(Some(next_run))
300        }
301        /// Start the scheduler
302        pub fn start(&self) -> Result<()> {
303            *self.running.lock().expect("Operation failed") = true;
304            Ok(())
305        }
306        /// Stop the scheduler
307        pub fn stop(&self) {
308            *self.running.lock().expect("Operation failed") = false;
309        }
310    }
311}
scirs2_io/workflow/functions.rs

scirs2_io/workflow/
functions.rs
