oxigdal_workflow/integrations/
mod.rs

1//! External workflow system integrations.
2//!
3//! Provides integration with popular workflow orchestration platforms:
4//! - Apache Airflow
5//! - Prefect
6//! - Temporal.io
7//! - Webhooks
8//! - Message queues (Kafka, RabbitMQ)
9
10pub mod external;
11
12#[cfg(feature = "integrations")]
13pub mod airflow;
14#[cfg(feature = "integrations")]
15pub mod prefect;
16#[cfg(feature = "integrations")]
17pub mod temporal;
18
19use crate::engine::WorkflowDefinition;
20use crate::error::{Result, WorkflowError};
21use serde::{Deserialize, Serialize};
22use std::collections::HashMap;
23
24#[cfg(feature = "integrations")]
25pub use airflow::AirflowIntegration;
26#[cfg(feature = "integrations")]
27pub use prefect::PrefectIntegration;
28#[cfg(feature = "integrations")]
29pub use temporal::TemporalIntegration;
30
31/// Integration type enumeration.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33pub enum IntegrationType {
34    /// Apache Airflow.
35    Airflow,
36    /// Prefect.
37    Prefect,
38    /// Temporal.io.
39    Temporal,
40    /// Webhook.
41    Webhook,
42    /// Kafka message queue.
43    Kafka,
44    /// RabbitMQ message queue.
45    RabbitMq,
46}
47
48/// Integration configuration.
49#[derive(Debug, Clone, Serialize, Deserialize)]
50pub struct IntegrationConfig {
51    /// Integration type.
52    pub integration_type: IntegrationType,
53    /// Endpoint URL.
54    pub endpoint: String,
55    /// Authentication credentials.
56    pub auth: Option<AuthConfig>,
57    /// Additional configuration.
58    pub extra_config: HashMap<String, String>,
59}
60
61/// Authentication configuration.
62#[derive(Debug, Clone, Serialize, Deserialize)]
63pub enum AuthConfig {
64    /// API key authentication.
65    ApiKey {
66        /// API key.
67        key: String,
68    },
69    /// Basic authentication.
70    Basic {
71        /// Username.
72        username: String,
73        /// Password.
74        password: String,
75    },
76    /// OAuth2 authentication.
77    OAuth2 {
78        /// Access token.
79        token: String,
80    },
81    /// None (no authentication).
82    None,
83}
84
85/// Webhook configuration.
86#[derive(Debug, Clone, Serialize, Deserialize)]
87pub struct WebhookConfig {
88    /// Webhook URL.
89    pub url: String,
90    /// HTTP method.
91    pub method: HttpMethod,
92    /// Headers to include.
93    pub headers: HashMap<String, String>,
94    /// Authentication.
95    pub auth: Option<AuthConfig>,
96    /// Retry configuration.
97    pub retry: Option<RetryConfig>,
98}
99
100/// HTTP method enumeration.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
102pub enum HttpMethod {
103    /// GET method.
104    Get,
105    /// POST method.
106    Post,
107    /// PUT method.
108    Put,
109    /// DELETE method.
110    Delete,
111    /// PATCH method.
112    Patch,
113}
114
115/// Retry configuration.
116#[derive(Debug, Clone, Serialize, Deserialize)]
117pub struct RetryConfig {
118    /// Maximum number of retries.
119    pub max_retries: usize,
120    /// Initial retry delay in milliseconds.
121    pub initial_delay_ms: u64,
122    /// Maximum retry delay in milliseconds.
123    pub max_delay_ms: u64,
124    /// Backoff multiplier.
125    pub backoff_multiplier: f64,
126}
127
128impl Default for RetryConfig {
129    fn default() -> Self {
130        Self {
131            max_retries: 3,
132            initial_delay_ms: 1000,
133            max_delay_ms: 30000,
134            backoff_multiplier: 2.0,
135        }
136    }
137}
138
139/// Integration manager for external systems.
140pub struct IntegrationManager {
141    configs: HashMap<String, IntegrationConfig>,
142}
143
144impl IntegrationManager {
145    /// Create a new integration manager.
146    pub fn new() -> Self {
147        Self {
148            configs: HashMap::new(),
149        }
150    }
151
152    /// Register an integration.
153    pub fn register(&mut self, name: String, config: IntegrationConfig) {
154        self.configs.insert(name, config);
155    }
156
157    /// Get an integration configuration.
158    pub fn get(&self, name: &str) -> Option<&IntegrationConfig> {
159        self.configs.get(name)
160    }
161
162    /// Remove an integration.
163    pub fn remove(&mut self, name: &str) -> Option<IntegrationConfig> {
164        self.configs.remove(name)
165    }
166
167    /// List all integrations.
168    pub fn list(&self) -> Vec<String> {
169        self.configs.keys().cloned().collect()
170    }
171
172    /// Export workflow to external format.
173    #[cfg_attr(not(feature = "integrations"), allow(unused_variables))]
174    pub fn export_workflow(
175        &self,
176        workflow: &WorkflowDefinition,
177        integration_type: IntegrationType,
178    ) -> Result<String> {
179        match integration_type {
180            #[cfg(feature = "integrations")]
181            IntegrationType::Airflow => AirflowIntegration::export_workflow(workflow),
182            #[cfg(feature = "integrations")]
183            IntegrationType::Prefect => PrefectIntegration::export_workflow(workflow),
184            #[cfg(feature = "integrations")]
185            IntegrationType::Temporal => TemporalIntegration::export_workflow(workflow),
186            _ => Err(WorkflowError::integration(
187                integration_type.as_str(),
188                "Export not implemented for this integration type",
189            )),
190        }
191    }
192
193    /// Trigger workflow via webhook.
194    #[cfg(feature = "integrations")]
195    pub async fn trigger_webhook(
196        &self,
197        config: &WebhookConfig,
198        payload: &serde_json::Value,
199    ) -> Result<String> {
200        use reqwest::Client;
201
202        let client = Client::new();
203        let mut request = match config.method {
204            HttpMethod::Get => client.get(&config.url),
205            HttpMethod::Post => client.post(&config.url),
206            HttpMethod::Put => client.put(&config.url),
207            HttpMethod::Delete => client.delete(&config.url),
208            HttpMethod::Patch => client.patch(&config.url),
209        };
210
211        // Add headers
212        for (key, value) in &config.headers {
213            request = request.header(key, value);
214        }
215
216        // Add authentication
217        if let Some(auth) = &config.auth {
218            request = match auth {
219                AuthConfig::ApiKey { key } => request.header("X-API-Key", key),
220                AuthConfig::Basic { username, password } => {
221                    request.basic_auth(username, Some(password))
222                }
223                AuthConfig::OAuth2 { token } => request.bearer_auth(token),
224                AuthConfig::None => request,
225            };
226        }
227
228        // Send request
229        let response =
230            request.json(payload).send().await.map_err(|e| {
231                WorkflowError::integration("webhook", format!("Request failed: {}", e))
232            })?;
233
234        let status = response.status();
235        let body = response.text().await.map_err(|e| {
236            WorkflowError::integration("webhook", format!("Failed to read response: {}", e))
237        })?;
238
239        if !status.is_success() {
240            return Err(WorkflowError::integration(
241                "webhook",
242                format!("Request failed with status {}: {}", status, body),
243            ));
244        }
245
246        Ok(body)
247    }
248}
249
250impl Default for IntegrationManager {
251    fn default() -> Self {
252        Self::new()
253    }
254}
255
256impl IntegrationType {
257    /// Get string representation.
258    pub fn as_str(&self) -> &'static str {
259        match self {
260            Self::Airflow => "airflow",
261            Self::Prefect => "prefect",
262            Self::Temporal => "temporal",
263            Self::Webhook => "webhook",
264            Self::Kafka => "kafka",
265            Self::RabbitMq => "rabbitmq",
266        }
267    }
268}
269
270#[cfg(test)]
271mod tests {
272    use super::*;
273
274    #[test]
275    fn test_integration_manager() {
276        let mut manager = IntegrationManager::new();
277
278        let config = IntegrationConfig {
279            integration_type: IntegrationType::Webhook,
280            endpoint: "https://example.com/webhook".to_string(),
281            auth: Some(AuthConfig::ApiKey {
282                key: "test-key".to_string(),
283            }),
284            extra_config: HashMap::new(),
285        };
286
287        manager.register("test-integration".to_string(), config);
288
289        assert!(manager.get("test-integration").is_some());
290        assert_eq!(manager.list().len(), 1);
291    }
292
293    #[test]
294    fn test_integration_type_str() {
295        assert_eq!(IntegrationType::Airflow.as_str(), "airflow");
296        assert_eq!(IntegrationType::Webhook.as_str(), "webhook");
297    }
298
299    #[test]
300    fn test_retry_config_default() {
301        let config = RetryConfig::default();
302        assert_eq!(config.max_retries, 3);
303        assert_eq!(config.initial_delay_ms, 1000);
304    }
305}
oxigdal_workflow/integrations/mod.rs

oxigdal_workflow/integrations/
mod.rs
