Scaler ses automatisations en production

Section 11.3.19 : Scaler ses automatisations en production

🎯 Objectif pédagogique

Apprendre à passer d'automatisations personnelles à des systèmes robustes et scalables. Vous serez capable de gérer les limites de rate, la résilience, et l'architecture multi-workflow.

De 5 automatisations à 50 — Les défis du scale

Marc a 5 scénarios Make, 3 scripts Python, et son Job Tracker. Tout fonctionne pour un seul utilisateur. Mais que se passe-t-il quand il doit automatiser des processus pour 10 collègues ? Ou quand il lance sa startup et a besoin de traiter 1000 emails/jour ?

Gestion des limites de rate (Rate Limiting)

# rate_limiter.py — Respecter les limites des APIs
import time
from functools import wraps
from collections import deque
from datetime import datetime, timedelta

class RateLimiter:
    """Contrôle le débit d'appels API."""
    
    def __init__(self, max_calls, period_seconds):
        self.max_calls = max_calls
        self.period = period_seconds
        self.calls = deque()
    
    def wait_if_needed(self):
        """Attendre si la limite est atteinte."""
        now = datetime.now()
        
        # Supprimer les appels hors de la fenêtre
        while self.calls and (now - self.calls[0]) > timedelta(seconds=self.period):
            self.calls.popleft()
        
        # Si limite atteinte, attendre
        if len(self.calls) >= self.max_calls:
            oldest = self.calls[0]
            wait_time = (oldest + timedelta(seconds=self.period) - now).total_seconds()
            if wait_time > 0:
                print(f"Rate limit: waiting \{wait_time:.1f\}s...")
                time.sleep(wait_time)
        
        self.calls.append(datetime.now())

# Utilisation
openai_limiter = RateLimiter(max_calls=60, period_seconds=60)  # 60 calls/min

def call_openai(prompt):
    """Appeler OpenAI en respectant le rate limit."""
    openai_limiter.wait_if_needed()
    # ... appel API ...

Pattern Retry avec Exponential Backoff

import time
import random

def retry_with_backoff(func, max_retries=3, base_delay=1):
    """Réessayer avec un délai exponentiel en cas d'erreur."""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise  # Dernière tentative, on lève l'erreur
            
            # Exponential backoff + jitter
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt \{attempt + 1\} failed: \{e\}. Retrying in \{delay:.1f\}s...")
            time.sleep(delay)

# Utilisation
result = retry_with_backoff(
    lambda: requests.get("https://api.example.com/data", timeout=10).json()
)

Architecture multi-workflow

Mauvais : 1 gros workflow monolithique
────────────────────────────────────────
[Trigger] → [API1] → [LLM] → [Parse] → [Router] → [DB] → [Email] → [Slack] → [Calendar]
→ Si un module échoue, TOUT s'arrête

Bon : Plusieurs petits workflows modulaires
────────────────────────────────────────
Workflow 1 : Collecte
[Schedule] → [API Jobs] → [Data Store : raw_jobs]

Workflow 2 : Analyse (déclenché par Workflow 1)
[Data Store trigger] → [LLM Classify] → [Data Store : classified_jobs]

Workflow 3 : Actions (déclenché par Workflow 2)
[Data Store trigger] → [Router]
  ├── High priority → [Slack + Email]
  ├── Medium → [Sheet only]
  └── Low → [Archive]

→ Chaque workflow est indépendant, testable, réparable séparément

File d'attente (Queue) pour les gros volumes

# Pour les traitements volumineux : file d'attente
import json
from pathlib import Path

class SimpleQueue:
    """File d'attente basée sur des fichiers JSON (sans Redis/RabbitMQ)."""
    
    def __init__(self, queue_dir="data/queue"):
        self.queue_dir = Path(queue_dir)
        self.queue_dir.mkdir(parents=True, exist_ok=True)
    
    def enqueue(self, item):
        """Ajouter un élément à la file."""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S_%f')
        filepath = self.queue_dir / f"\{timestamp\}.json"
        filepath.write_text(json.dumps(item, ensure_ascii=False), encoding='utf-8')
    
    def dequeue(self):
        """Retirer et retourner le plus ancien élément."""
        files = sorted(self.queue_dir.glob("*.json"))
        if not files:
            return None
        
        filepath = files[0]
        item = json.loads(filepath.read_text(encoding='utf-8'))
        filepath.unlink()  # Supprimer après traitement
        return item
    
    def size(self):
        """Nombre d'éléments en attente."""
        return len(list(self.queue_dir.glob("*.json")))

# Utilisation :
queue = SimpleQueue()

# Producteur (collecte des emails)
for email in new_emails:
    queue.enqueue(\{"email_id": email.id, "subject": email.subject\})

# Consommateur (traitement par l'IA)
while queue.size() > 0:
    item = queue.dequeue()
    result = analyze_with_ai(item)
    save_result(result)

🏋️ Exercice pratique (25 minutes)

1. →Ajoutez le RateLimiter à votre script d'appel OpenAI
2. →Ajoutez le retry_with_backoff à vos appels HTTP
3. →Testez en simulant des échecs (timeout, serveur indisponible)
4. →Refactorisez un workflow Make monolithique en 2-3 workflows modulaires