We making IT digital 4 U

01# Automatisation Niveau 5 - Pipeline n8n + Python
02import requests
03from datetime import datetime
04
05class AutomationEngine:
06    def __init__(self, api_key, webhook_url):
07        self.api_key = api_key
08        self.webhook_url = webhook_url
09        self.session = requests.Session()
10
11    def trigger_workflow(self, data):
12        """Déclenche un workflow n8n avec monitoring"""
13        headers = {
14            'Authorization': f'Bearer {self.api_key}',
15            'Content-Type': 'application/json'
16        }
17
18        payload = {
19            'timestamp': datetime.utcnow().isoformat(),
20            'data': data,
21            'environment': 'production'
22        }
23
24        response = self.session.post(
25            self.webhook_url,
26            json=payload,
27            headers=headers,
28            timeout=30
29        )
30
31        if response.status_code == 200:
32            return {'status': 'success', 'data': response.json()}
33        else:
34            self._handle_error(response)
35
36    def _handle_error(self, response):
37        # Logging + alertes Slack/Discord
38        error_data = {
39            'code': response.status_code,
40            'message': response.text,
41            'timestamp': datetime.utcnow()
42        }
43        # Envoi vers système de monitoring
44        self._send_alert(error_data)
45
46# Infrastructure self-hosted, scalable, monitorée