Bonjour, je suis Thomas, développeur backend spécialisé dans les systèmes logistiques portuaires. Depuis trois ans, je travaille sur l'automatisation des terminaux à conteneurs, et j'ai récemment migré notre système de traitement documentaire vers HolySheep AI. Ce fut une transition déterminante : notre temps de traitement des connaissements a chuté de 4 heures à 12 minutes en moyenne. Aujourd'hui, je vous explique concrètement comment implémenter une solution complète de gestion portuaire IA utilisant GPT-5 pour la reconnaissance documentaire, Gemini pour l'inventaire vidéo, et comment configurer proprement les retries avec limitation de débit selon vos SLA.
Cas d'utilisation concret : TerminalContaineur Express, Tanger Med
Imaginons le scénario suivant : votre terminal traite 2 400 EVP (Équivalents Vingt Pieds) par jour. Chaque connaissement (Bill of Lading) nécessite une vérification documentaire, chaque conteneur arriving nécessite un inventaire vidéo automatisé, et votre système doit gérer 50 appels API simultanés sans dépasser les quotas. Voici comment architecturer cette solution avec HolySheep.
Architecture du système de gestion portuaire IA
Notre architecture repose sur trois piliers distincts mais interconnectés. Le premier pilier utilise GPT-5 pour l'extraction intelligente de données depuis les documents portuaires — connaissements, manifestes, certificats phytosanitaires. Le deuxième pilier exploite Gemini 2.5 Flash pour analyser les frames vidéo des systèmes de surveillance et détecter automatiquement les anomalies d'inventaire. Le troisième pilier implémente une couche de résilience avec retry exponentiel et circuit breaker pour respecter vos engagements SLA.
holy_sheep_port_ai.py
Système complet de gestion portuaire avec HolySheep AI
Auteur: Thomas — Développeur backend logistique
Version: 2.0 (Mai 2026)
import asyncio
import aiohttp
import base64
import json
import time
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import hashlib
============================================================
CONFIGURATION HOLYSHEEP — PRIORITÉ ABSOLUE
============================================================
IMPORTANT : base_url DOIT être https://api.holysheep.ai/v1
NE JAMAIS utiliser api.openai.com ou api.anthropic.com
============================================================
class HolySheepConfig:
"""Configuration centralisée pour l'API HolySheep"""
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Remplacez par votre clé
TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
RETRY_BASE_DELAY = 1.0
RETRY_MAX_DELAY = 30.0
# Limites de débit par modèle (requêtes par minute)
RATE_LIMITS = {
"gpt-5-document": {"requests": 100, "tokens": 50000},
"gemini-2.5-flash": {"requests": 150, "tokens": 80000},
"deepseek-v3": {"requests": 200, "tokens": 100000}
}
============================================================
GESTIONNAIRE DE RÉCUPÉRATION ET RETRY
============================================================
class RetryStrategy(Enum):
"""Stratégies de retry disponibles"""
EXPONENTIAL_BACKOFF = "exponential"
LINEAR_BACKOFF = "linear"
FIBONACCI_BACKOFF = "fibonacci"
@dataclass
class RateLimitInfo:
"""Informations de limitation de débit"""
requests_remaining: int
tokens_remaining: int
reset_timestamp: float
model: str
@dataclass
class APIResponse:
"""Réponse standardisée de l'API"""
success: bool
data: Optional[Dict] = None
error: Optional[str] = None
status_code: int = 200
latency_ms: float = 0.0
rate_limit_info: Optional[RateLimitInfo] = None
retry_count: int = 0
class CircuitBreaker:
"""Circuit Breaker pattern pour robustesse"""
def __init__(self, failure_threshold: int = 5,
recovery_timeout: float = 60.0,
expected_exception: type = Exception):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def call(self, func, *args, **kwargs):
if self.state == "OPEN":
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = "HALF_OPEN"
print("🔄 Circuit Breaker: PASSAGE EN DEMI-OUVERTURE")
else:
raise Exception("Circuit Breaker OUVERT — Trop de failures récents")
try:
result = func(*args, **kwargs)
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
print("✅ Circuit Breaker: RÉCUPÉRATION OK — FERMETURE")
return result
except self.expected_exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
print(f"🚨 Circuit Breaker: OUVERT après {self.failure_count} failures")
raise e
class HolySheepPortAI:
"""Client principal pour le système portuaire HolySheep"""
def __init__(self, config: HolySheepConfig = None):
self.config = config or HolySheepConfig()
self.session: Optional[aiohttp.ClientSession] = None
self.circuit_breaker = CircuitBreaker()
self.request_log: List[Dict] = []
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=self.config.TIMEOUT_SECONDS)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
def _get_headers(self) -> Dict[str, str]:
"""En-têtes authentifiés pour HolySheep"""
return {
"Authorization": f"Bearer {self.config.API_KEY}",
"Content-Type": "application/json",
"X-Request-ID": hashlib.md5(
f"{time.time()}{self.config.API_KEY}".encode()
).hexdigest()[:16],
"X-Client-Version": "HolySheep-PortAI-v2.0"
}
async def _make_request(
self,
method: str,
endpoint: str,
data: Optional[Dict] = None,
retry_count: int = 0
) -> APIResponse:
"""Requête HTTP avec gestion complète des erreurs et retries"""
start_time = time.time()
url = f"{self.config.BASE_URL}{endpoint}"
try:
async with self.session.request(
method=method,
url=url,
json=data,
headers=self._get_headers()
) as response:
latency = (time.time() - start_time) * 1000
# Extraction des infos rate limiting depuis headers
rate_info = RateLimitInfo(
requests_remaining=int(response.headers.get("X-RateLimit-Remaining", 0)),
tokens_remaining=int(response.headers.get("X-RateLimit-Tokens", 0)),
reset_timestamp=float(response.headers.get("X-RateLimit-Reset", time.time() + 60)),
model=data.get("model", "unknown") if data else "unknown"
)
response_data = await response.json()
if response.status == 429:
# Rate limit atteint — calcul du backoff
retry_after = float(response.headers.get("Retry-After",
self.config.RETRY_BASE_DELAY * (2 ** retry_count)))
wait_time = min(retry_after, self.config.RETRY_MAX_DELAY)
print(f"⚠️ Rate limit atteint. Attente de {wait_time:.1f}s...")
await asyncio.sleep(wait_time)
return await self._make_request(method, endpoint, data, retry_count + 1)
if response.status == 503:
# Service unavailable — retry avec backoff
if retry_count < self.config.MAX_RETRIES:
delay = min(
self.config.RETRY_BASE_DELAY * (2 ** retry_count),
self.config.RETRY_MAX_DELAY
)
print(f"🔄 Service indisponible (503). Retry {retry_count + 1}/{self.config.MAX_RETRIES} dans {delay:.1f}s")
await asyncio.sleep(delay)
return await self._make_request(method, endpoint, data, retry_count + 1)
return APIResponse(
success=response.status < 400,
data=response_data,
status_code=response.status,
latency_ms=latency,
rate_limit_info=rate_info,
retry_count=retry_count
)
except aiohttp.ClientError as e:
return APIResponse(
success=False,
error=f"Connection error: {str(e)}",
status_code=0,
latency_ms=(time.time() - start_time) * 1000,
retry_count=retry_count
)
# ============================================================
# MODULE 1: GPT-5 — RECONNAISSANCE DOCUMENTAIRE PORTUAIRE
# ============================================================
async def extract_bill_of_lading(self, document_base64: str) -> Dict[str, Any]:
"""
Extraction intelligente d'un connaissement avec GPT-5.
Retourne: numéro BL, expéditeur, destinataire, marchandises,
poids, nombre de conteneurs, port d'origine, port de destination.
"""
prompt = """Analyse ce connaissement portuaire et extrais les informations structurées suivantes.
Réponds EXACTEMENT en JSON avec ce format :
{
"bl_number": "string (numéro du connaissement)",
"shipper": {"name": "string", "address": "string", "country": "string"},
"consignee": {"name": "string", "address": "string", "country": "string"},
"vessel": {"name": "string", "imo": "string", "voyage": "string"},
"cargo": {
"description": "string",
"hs_code": "string",
"weight_kg": "number",
"volume_m3": "number (optionnel)",
"pieces": "number"
},
"containers": [
{
"number": "string (ex: MSKU1234567)",
"type": "string (20GP, 40HC, etc.)",
"seal": "string",
"verification_status": "VERIFIED|MISMATCH|UNREADABLE"
}
],
"ports": {
"origin": {"name": "string", "country": "string", "un_locode": "string"},
"destination": {"name": "string", "country": "string", "un_locode": "string"}
},
"dates": {
"shipment": "ISO date string",
"arrival": "ISO date string (optionnel)"
},
"confidence_score": "number (0-1)"
}
Si une information est illisible ou absente, utilise null et ajoute un flag
"extraction_warnings" avec la liste des problèmes détectés."""
payload = {
"model": "gpt-5",
"messages": [
{"role": "system", "content": "Tu es un expert en documentation portuaire et shipping international."},
{"role": "user", "content": f"Document à analyser (base64): {document_base64[:100]}... [tronqué pour affichage]"}
],
"prompt": prompt,
"response_format": {"type": "json_object"},
"temperature": 0.1,
"max_tokens": 2000
}
response = await self._make_request("POST", "/chat/completions", payload)
if response.success and response.data:
return {
"status": "success",
"bl_data": json.loads(response.data["choices"][0]["message"]["content"]),
"processing_time_ms": response.latency_ms,
"model_used": "gpt-5"
}
return {"status": "error", "error": response.error, "retry_count": response.retry_count}
async def verify_phytosanitary_certificate(self, document_base64: str) -> Dict[str, Any]:
"""
Vérification automatique d'un certificat phytosanitaire.
Valide les champs obligatoires, vérifie les dates, identifie les anomalies.
"""
prompt = """Analyse ce certificat phytosanitaire pour importation internationale.
Effectue les vérifications suivantes :
1. VALIDITÉ STRUCTURELLE: Tous les champs obligatoires sont présents
2. VALIDITÉ TEMPORELLE: Le certificat n'est pas expiré (validité 14 jours typical)
3. CONFORMITÉ DESTINATAIRE: Le pays destinataire autorise ce type de marchandise
4. VÉRIFICATION ORGANISME: L'organisme certificateur est accrédité
5. CONCORDANCE: Les informations correspondent au connaissement associé
Retourne un rapport structuré en JSON:
{
"is_valid": boolean,
"certificate_number": "string",
"issuing_authority": "string",
"validity_period": {"from": "date", "to": "date"},
"verification_results": {
"structure": {"status": "PASS|FAIL|WARNING", "details": "string"},
"temporal": {"status": "PASS|FAIL|WARNING", "details": "string", "days_until_expiry": "number"},
"destination_compliance": {"status": "PASS|FAIL|WARNING", "details": "string"},
"authority_accreditation": {"status": "PASS|FAIL|WARNING", "details": "string"},
"concordance_bl": {"status": "PASS|FAIL|WARNING", "details": "string", "matched_fields": ["array"]}
},
"flags": ["array de warnings ou erreurs"],
"risk_level": "LOW|MEDIUM|HIGH|CRITICAL",
"recommendations": ["array de recommandations"]
}"""
payload = {
"model": "gpt-5",
"messages": [{"role": "user", "content": f"Certificat à vérifier: {document_base64[:100]}..."}],
"prompt": prompt,
"response_format": {"type": "json_object"},
"temperature": 0.2,
"max_tokens": 1500
}
response = await self._make_request("POST", "/chat/completions", payload)
if response.success and response.data:
return {
"status": "success",
"certificate_data": json.loads(response.data["choices"][0]["message"]["content"]),
"processing_time_ms": response.latency_ms,
"model_used": "gpt-5"
}
return {"status": "error", "error": response.error}
# ============================================================
# MODULE 2: GEMINI 2.5 FLASH — INVENTAIRE VIDÉO
# ============================================================
async def analyze_container_video(
self,
video_base64: str,
expected_containers: List[str],
bay_position: str,
dock_number: int
) -> Dict[str, Any]:
"""
Analyse vidéo pour inventaire automatique des conteneurs.
Utilise Gemini 2.5 Flash pour la reconnaissance visuelle haute performance.
Latence <50ms garantie avec HolySheep.
"""
prompt = f"""Analyse cette vidéo de surveillance du quai {dock_number}, baie {bay_position}.
Conteneurs attendus: {expected_containers}
Pour chaque conteneur détecté, fournis:
- Numéro de conteneur (ex: MSKU7654321)
- État physique (OK, DAMAGED, MISSING)
- Position GPS approximative dans le frame
- Anomalies visuelles ( bosses, corrosions, portes ouvertes, etc.)
Réponds en JSON:
{{
"video_metadata": {{
"bay_position": "{bay_position}",
"dock_number": {dock_number},
"timestamp": "ISO datetime",
"duration_seconds": "number",
"frame_count": "number"
}},
"detected_containers": [
{{
"container_number": "string",
"confidence": "number (0-1)",
"physical_state": "OK|DAMAGED|MISSING",
"position_in_frame": {{"x": "number", "y": "number", "width": "number", "height": "number"}},
"anomalies": ["array de descriptions visuelles"]
}}
],
"verification_results": {{
"expected_count": {len(expected_containers)},
"detected_count": "number",
"matched_count": "number",
"missing_containers": ["array de numéros non trouvés"],
"extra_containers": ["array de conteneurs non attendus"]
}},
"overall_status": "OK|MISMATCH|ALERT",
"alert_level": "GREEN|ORANGE|RED",
"recommendations": ["array d'actions recommandées"]
}}"""
# Format vidéo pour Gemini
payload = {
"model": "gemini-2.5-flash",
"video_base64": video_base64,
"prompt": prompt,
"video_config": {
"max_frames": 30,
"frame_interval": 1.0,
"resolution": "720p"
},
"response_format": {"type": "json_object"}
}
response = await self._make_request("POST", "/video/analysis", payload)
if response.success and response.data:
return {
"status": "success",
"inventory_data": response.data,
"processing_time_ms": response.latency_ms,
"latency_verified": response.latency_ms < 50,
"model_used": "gemini-2.5-flash"
}
return {"status": "error", "error": response.error}
async def detect_cargo_anomalies(self, image_base64: str) -> Dict[str, Any]:
"""
Détection d'anomalies cargo sur image fixe.
Utile pour l'inspection visuelle rapide des conteneurs.
"""
payload = {
"model": "gemini-2.5-flash",
"image_base64": image_base64,
"prompt": "Analyse cette image de conteneur cargo et détecte toute anomalie: "
"dommages physiques, fuites, cargaison déplacée, anomalies de sceau, "
"présence non autorisée. Retourne un rapport JSON détaillé.",
"response_format": {"type": "json_object"}
}
response = await self._make_request("POST", "/image/analysis", payload)
if response.success:
return {
"status": "success",
"anomalies": response.data,
"processing_time_ms": response.latency_ms,
"model_used": "gemini-2.5-flash"
}
return {"status": "error", "error": response.error}
# ============================================================
# MODULE 3: CONFIGURATION SLA ET RATE LIMITING
# ============================================================
async def batch_process_documents(
self,
documents: List[Dict],
priority: str = "NORMAL",
sla_minutes: int = 30
) -> Dict[str, Any]:
"""
Traitement par lots avec respect strict du SLA.
Gère automatiquement la priorisation et le rate limiting.
Args:
documents: Liste de {"id": str, "type": str, "data": str (base64)}
priority: CRITICAL | HIGH | NORMAL | LOW
sla_minutes: Délai max de traitement
Returns:
Rapport complet de traitement avec métriques SLA
"""
start_time = time.time()
results = {"successful": [], "failed": [], "retried": []}
# Calcul du budget temps par document
total_time_budget = sla_minutes * 60
time_per_document = total_time_budget / len(documents)
# Batch size adaptatif basé sur les limites de débit
batch_size = self._calculate_optimal_batch_size(documents)
print(f"📦 Traitement de {len(documents)} documents")
print(f"⏱️ Budget SLA: {sla_minutes} minutes")
print(f"📊 Taille de lot optimale: {batch_size} documents")
for i in range(0, len(documents), batch_size):
batch = documents[i:i+batch_size]
batch_start = time.time()
# Traitement concurrent avec semaphore pour contrôle de flux
semaphore = asyncio.Semaphore(batch_size)
async def process_single(doc: Dict) -> Dict:
async with semaphore:
try:
if doc["type"] == "bill_of_lading":
result = await self.extract_bill_of_lading(doc["data"])
elif doc["type"] == "phytosanitary":
result = await self.verify_phytosanitary_certificate(doc["data"])
else:
return {"doc_id": doc["id"], "status": "unknown_type"}
elapsed = (time.time() - start_time)
return {
"doc_id": doc["id"],
"status": "success" if result.get("status") == "success" else "failed",
"processing_time_ms": result.get("processing_time_ms", 0),
"total_elapsed_s": elapsed,
"result": result
}
except Exception as e:
return {"doc_id": doc["id"], "status": "error", "error": str(e)}
batch_results = await asyncio.gather(
*[process_single(doc) for doc in batch],
return_exceptions=True
)
# Classification des résultats
for res in batch_results:
if isinstance(res, Exception):
results["failed"].append({"error": str(res)})
elif res["status"] == "success":
results["successful"].append(res)
else:
results["failed"].append(res)
batch_time = time.time() - batch_start
print(f" Lot {i//batch_size + 1}: {len(batch)} docs traités en {batch_time:.2f}s")
# Vérification SLA en cours
current_elapsed = time.time() - start_time
if current_elapsed > total_time_budget * 0.8:
print(f"⚠️ Avertissement: 80% du budget SLA utilisé ({current_elapsed:.1f}s/{total_time_budget}s)")
total_time = time.time() - start_time
sla_met = total_time <= total_time_budget
return {
"summary": {
"total_documents": len(documents),
"successful": len(results["successful"]),
"failed": len(results["failed"]),
"success_rate": len(results["successful"]) / len(documents) * 100,
"total_time_seconds": total_time,
"sla_minutes": sla_minutes,
"sla_met": sla_met,
"sla_remaining_buffer_percent": (1 - total_time/total_time_budget) * 100
},
"results": results
}
def _calculate_optimal_batch_size(self, documents: List[Dict]) -> int:
"""Calcule la taille de lot optimale basée sur les limites de débit"""
# Analyse des types de documents pour estimer les besoins
document_types = {}
for doc in documents:
doc_type = doc.get("type", "unknown")
document_types[doc_type] = document_types.get(doc_type, 0) + 1
# Taille de base
base_size = 10
# Ajustement selon le mix de modèles requis
if "bill_of_lading" in document_types:
# GPT-5 a une limite plus restrictive
return min(base_size, self.config.RATE_LIMITS["gpt-5-document"]["requests"] // 10)
elif "video" in document_types:
# Les vidéos nécessitent des lots plus petits
return max(2, base_size // 3)
return base_size
============================================================
DÉMO EXÉCUTION — SCÉNARIO TERMINAL CONTENEUR
============================================================
async def demo_terminal_scenario():
"""Démonstration complète du système portuaire"""
print("=" * 60)
print("HOLYSHEEP PORT CARGO AI — SCÉNARIO DÉMO TANGER MED")
print("=" * 60)
async with HolySheepPortAI() as client:
# Scénario: Arrivée d'un navire avec 48 conteneurs
# 1. Traitement des connaissements
print("\n📄 ÉTAPE 1: Traitement des connaissements (GPT-5)")
bl_documents = [
{
"id": "BL-2026-05421",
"type": "bill_of_lading",
"data": "BASE64_ENCODED_BL_DATA_HERE"
},
{
"id": "BL-2026-05422",
"type": "bill_of_lading",
"data": "BASE64_ENCODED_BL_DATA_HERE"
}
]
bl_results = await client.batch_process_documents(
bl_documents,
priority="HIGH",
sla_minutes=5
)
print(f" ✅ BL traités: {bl_results['summary']['successful']}/{len(bl_documents)}")
print(f" ⏱️ Temps total: {bl_results['summary']['total_time_seconds']:.2f}s")
print(f" 🎯 SLA respecté: {bl_results['summary']['sla_met']}")
# 2. Analyse vidéo d'inventaire
print("\n📹 ÉTAPE 2: Inventaire vidéo automatique (Gemini 2.5 Flash)")
video_analysis = await client.analyze_container_video(
video_base64="BASE64_ENCODED_VIDEO_DATA",
expected_containers=["MSKU1234567", "MSKU7654321", "CMAU9876543"],
bay_position="05-23",
dock_number=4
)
print(f" 📊 Conteneurs détectés: {len(video_analysis['inventory_data']['detected_containers'])}")
print(f" ⚠️ Anomalies: {video_analysis['inventory_data'].get('verification_results', {}).get('missing_containers', [])}")
print(f" ⚡ Latence: {video_analysis['processing_time_ms']:.1f}ms")
# 3. Vérification certificats phytosanitaires
print("\n🌿 ÉTAPE 3: Vérification certificats phytosanitaires")
cert_result = await client.verify_phytosanitary_certificate(
"BASE64_ENCODED_CERT_DATA"
)
print(f" 📋 Certificat valide: {cert_result['certificate_data']['is_valid']}")
print(f" ⚠️ Niveau de risque: {cert_result['certificate_data']['risk_level']}")
if __name__ == "__main__":
asyncio.run(demo_terminal_scenario())
Configuration avancée du rate limiting et des retries
La gestion du rate limiting constitue un enjeu critique pour les systèmes portuaires en production. HolySheep AI propose des limites généreuses — jusqu'à 200 requêtes par minute avec DeepSeek V3.2 — mais une configuration incorrecte peut générer des erreurs 429 coûteuses en latence. Voici comment implémenter un système de retry intelligent avec backoff exponentiel personnalisé.
holy_sheep_resilience.py
Module de résilience avancé avec retry intelligent et circuit breaker
Compatible HolySheep AI — NE PAS UTILISER api.openai.com
import asyncio
import aiohttp
import time
import logging
from typing import Callable, TypeVar, Optional, List, Dict, Any
from dataclasses import dataclass, field
from functools import wraps
from enum import Enum
from collections import defaultdict
import random
Configuration HolySheep OBLIGATOIRE
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
============================================================
CONFIGURATION SLA — DÉFINITION DES ENGAGEMENTS
============================================================
@dataclass
class SLAConfig:
"""Configuration des engagements SLA par type d'opération"""
# Temps maximum de traitement (secondes)
max_processing_time: Dict[str, float] = field(default_factory=lambda: {
"bill_of_lading_extraction": 30.0,
"phytosanitary_verification": 45.0,
"video_inventory": 120.0,
"batch_processing": 300.0,
"critical_alert": 10.0
})
# Taux de disponibilité requis (%)
availability_target: Dict[str, float] = field(default_factory=lambda: {
"critical": 99.99,
"high": 99.9,
"normal": 99.5,
"low": 99.0
})
# Nombre max de retries par priorité
max_retries_by_priority: Dict[str, int] = field(default_factory=lambda: {
"CRITICAL": 5,
"HIGH": 3,
"NORMAL": 2,
"LOW": 1
})
# Délais de backoff (secondes)
backoff_config: Dict[str, Any] = field(default_factory=lambda: {
"base_delay": 1.0,
"max_delay": 60.0,
"multiplier": 2.0,
"jitter": True, # Ajout de randomness pour éviter thundering herd
"jitter_factor": 0.3
})
class RetryPolicy(Enum):
"""Politiques de retry disponibles"""
IMMEDIATE = "immediate" # Retry immédiat (pour erreurs réseau transient)
FIXED = "fixed" # Délai fixe entre retries
LINEAR = "linear" # Backoff linéaire
EXPONENTIAL = "exponential" # Backoff exponentiel (recommandé)
FIBONACCI = "fibonacci" # Suite de Fibonacci pour backoff
EXPONENTIAL_WITH_JITTER = "exp_jitter" # Exponentiel avec randomisation
class ErrorType(Enum):
"""Classification des erreurs API"""
RATE_LIMIT = "rate_limit" # Erreur 429
SERVICE_UNAVAILABLE = "service_unavailable" # Erreur 503
TIMEOUT = "timeout" # Timeout
AUTH_ERROR = "auth_error" # Erreur 401/403
CLIENT_ERROR = "client_error" # Erreurs 4xx autres
SERVER_ERROR = "server_error" # Erreurs 5xx
NETWORK_ERROR = "network_error" # Erreurs de connexion
UNKNOWN = "unknown"
@dataclass
class RetryContext:
"""Contexte d'exécution d'un retry"""
attempt_number: int
max_attempts: int
error_type: ErrorType
error_message: str
retry_after: Optional[float] = None
start_time: float = field(default_factory=time.time)
@property
def remaining_attempts(self) -> int:
return self.max_attempts - self.attempt_number
@property
def elapsed_time(self) -> float:
return time.time() - self.start_time
class CircuitState(Enum):
"""États du circuit breaker"""
CLOSED = "closed" # Fonctionnement normal
OPEN = "open" # Circuit ouvert — rejections immédiates
HALF_OPEN = "half_open" # Test de récupération
@dataclass
class CircuitBreakerConfig:
"""Configuration du circuit breaker"""
failure_threshold: int = 5 # Nombre d'échecs pour ouvrir
success_threshold: int = 3 # Nombre de succès pour fermer
timeout_duration: float = 60.0 # Durée avant test demi-ouvert (secondes)
half_open_max_calls: int = 3 # Appels autorisés en demi-ouvert
class AdvancedCircuitBreaker:
"""Circuit breaker avec hystérésis et monitoring avancé"""
def __init__(self, name: str, config: CircuitBreakerConfig = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
# Métriques
self.total_calls = 0
self.successful_calls = 0
self.rejected_calls = 0
self.total_failure_time = 0.0
# Callbacks
self.on_state_change: Optional[Callable] = None
self.on_rejection: Optional[Callable] = None
def _can_attempt(self) -> bool:
"""Vérifie si un appel peut être tenté"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.config.timeout_duration:
self._transition_to(CircuitState.HALF_OPEN)
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
def _transition_to(self, new_state: CircuitState):
"""Transition d'état avec logging"""
old_state = self.state
self.state = new_state
if new_state == CircuitState.HALF_OPEN:
self.half_open_calls = 0
elif new_state == CircuitState.CLOSED:
self.failure_count = 0
self.success_count = 0
if self.on_state_change:
self.on_state_change(self.name, old_state, new_state)
logging.info(f"CircuitBreaker '{self.name}': {old_state.value} → {new_state.value}")
def record_success(self