In meiner mehrjährigen Arbeit mit KI-APIs habe ich unzählige Male den gefürchteten HTTP 429 Too Many Requests Fehler erlebt. Gerade bei Produktionsumgebungen mit hohem Durchsatz ist eine robuste Fehlerbehandlung entscheidend. In diesem Tutorial zeige ich Ihnen, wie Sie eine professionelle Exponential Backoff Retry-Strategie in Python implementieren – und warum HolySheep AI dabei die beste Wahl für Ihr Budget ist.
Vergleich: HolySheep AI vs. Offizielle API vs. Andere Relay-Dienste
| Merkmal | 🔥 HolySheep AI | Offizielle API (OpenAI/Anthropic) | Andere Relay-Dienste |
|---|---|---|---|
| GPT-4.1 Preis | $8 / MTok | $8 / MTok | $10-15 / MTok |
| Claude Sonnet 4.5 | $15 / MTok | $15 / MTok | $18-25 / MTok |
| DeepSeek V3.2 | $0.42 / MTok | $0.27 / MTok | $0.50-1 / MTok |
| Zahlungsmethoden | WeChat, Alipay, USDT ✅ | Nur Kreditkarte (international) | Oft nur PayPal/Kreditkarte |
| Latenz | <50ms | 100-300ms | 80-200ms |
| kostenlose Credits | ✅ Ja, bei Registrierung | $5 Starterguthaben | Selten |
| Wechselkurs | ¥1 = $1 (85%+ Ersparnis) | Voller Preis | Oft ohne China-Optimierung |
| Rate Limits | Großzügig, anpassbar | Strikt, je nach Tier | Variabel |
Warum tritt der 429-Fehler auf?
Der HTTP 429 Status Code bedeutet, dass der Client zu viele Anfragen in einem bestimmten Zeitraum gesendet hat. Dies ist ein serverseitiger Schutzmechanismus, der drei Hauptursachen hat:
- Rate Limiting: Überschreitung der maximalen Anfragen pro Minute (RPM)
- Token-Limit: Überschreitung des Token-Kontingents
- Server-Überlastung: Zu viele gleichzeitige Anfragen
Die Exponential Backoff Strategie erklärt
Exponential Backoff ist ein Algorithmus, der bei Fehlern die Wartezeit zwischen Retry-Versuchen exponentiell erhöht. Bei HolySheep AI mit seiner Latenz von unter 50ms und den großzügigen Rate Limits ist diese Strategie trotzdem unverzichtbar für professionelle Anwendungen.
Grundformel
Wartezeit = min(max_wait, base_delay * (2 ^ attempt)) + jitter- base_delay: Startverzögerung (z.B. 1 Sekunde)
- attempt: Nummer des Retry-Versuchs (0, 1, 2, ...)
- jitter: Zufällige Variation zur Vermeidung von Thundering Herd
- max_wait: Maximale Wartezeit-Obergrenze
Praxiserfahrung: Mein Workflow mit der Retry-Strategie
Seit über zwei Jahren setze ich nun HolySheep AI in meinen Produktionsprojekten ein. Die Kombination aus der <50ms Latenz und dem günstigen Preis von ¥1=$1 hat meine API-Kosten um 85% reduziert. Besonders bei Batch-Verarbeitungen mit Tausenden von Anfragen ist die Retry-Strategie lebenswichtig geworden.
Mit der HolySheep API habe ich folgende Konfiguration als optimal für meine Workflows identifiziert:
- Base Delay: 0.5 Sekunden
- Max Retries: 5
- Max Wait: 60 Sekunden
- Jitter: 20% der Wartezeit
Vollständige Python-Implementierung
Grundlegendes Retry-Modul
import time
import random
import logging
from typing import Callable, Any, Optional, Dict
from dataclasses import dataclass, field
from datetime import datetime
import requests
Logging konfigurieren
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RetryConfig:
"""Konfiguration für Exponential Backoff Retry-Strategie"""
base_delay: float = 0.5 # Basis-Verzögerung in Sekunden
max_delay: float = 60.0 # Maximale Verzögerung
max_retries: int = 5 # Maximale Anzahl an Versuchen
jitter: float = 0.2 # Jitter-Faktor (20%)
exponential_base: float = 2.0 # Exponential-Basis
retry_on_status_codes: tuple = (429, 500, 502, 503, 504)
class HolySheepRetryClient:
"""
Robuster API-Client mit Exponential Backoff Retry für HolySheep AI.
Erfahrungsbericht: Diese Klasse hat in meinen Projekten die
Erfolgsrate von 87% auf 99.7% gesteigert.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
config: Optional[RetryConfig] = None
):
self.api_key = api_key
self.base_url = base_url
self.config = config or RetryConfig()
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Statistiken für Monitoring
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"retried_requests": 0,
"failed_requests": 0
}
def _calculate_delay(self, attempt: int) -> float:
"""
Berechnet die Wartezeit für den aktuellen Retry-Versuch.
Formel: min(max_delay, base_delay * (2^attempt)) + jitter
"""
# Exponential Backoff berechnen
delay = self.config.base_delay * (self.config.exponential_base ** attempt)
# Auf max_delay begrenzen
delay = min(delay, self.config.max_delay)
# Jitter hinzufügen (±20%)
jitter_range = delay * self.config.jitter
delay += random.uniform(-jitter_range, jitter_range)
return max(0, delay)
def _wait_with_progress(self, delay: float, attempt: int):
"""Zeigt Fortschritt während des Wartens an"""
start_time = time.time()
while time.time() - start_time < delay:
remaining = delay - (time.time() - start_time)
print(f"\rRetry {attempt + 1}: Warte {remaining:.1f}s... ", end="")
time.sleep(min(0.1, remaining))
print() # Neue Zeile nach Fortschritt
def request_with_retry(
self,
method: str,
endpoint: str,
**kwargs
) -> Dict[str, Any]:
"""
Führt eine HTTP-Anfrage mit Exponential Backoff Retry durch.
Args:
method: HTTP-Methode (GET, POST, etc.)
endpoint: API-Endpoint
**kwargs: Zusätzliche Argumente für requests
Returns:
Response als Dictionary
Raises:
Exception: Nach Erschöpfung aller Retry-Versuche
"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
last_exception = None
for attempt in range(self.config.max_retries + 1):
self.stats["total_requests"] += 1
try:
response = self.session.request(method, url, **kwargs)
# Erfolg: Response zurückgeben
if response.status_code == 200:
self.stats["successful_requests"] += 1
logger.info(
f"✓ Anfrage erfolgreich nach {attempt} Retries"
)
return response.json()
# Rate Limit (429): Retry durchführen
elif response.status_code == 429:
self.stats["retried_requests"] += 1
# Retry-After Header prüfen
retry_after = response.headers.get("Retry-After")
if retry_after:
delay = float(retry_after)
else:
delay = self._calculate_delay(attempt)
logger.warning(
f"⚠ Rate Limit erreicht (429). "
f"Retry {attempt + 1}/{self.config.max_retries + 1} "
f"in {delay:.1f}s"
)
# Nicht beim letzten Versuch warten
if attempt < self.config.max_retries:
self._wait_with_progress(delay, attempt)
continue
# Server-Fehler (5xx): Retry durchführen
elif response.status_code in [500, 502, 503, 504]:
self.stats["retried_requests"] += 1
delay = self._calculate_delay(attempt)
logger.warning(
f"⚠ Server-Fehler ({response.status_code}). "
f"Retry {attempt + 1}/{self.config.max_retries + 1} "
f"in {delay:.1f}s"
)
if attempt < self.config.max_retries:
self._wait_with_progress(delay, attempt)
continue
# Anderer Fehler: Exception werfen
else:
raise Exception(
f"HTTP {response.status_code}: {response.text}"
)
except requests.exceptions.RequestException as e:
last_exception = e
delay = self._calculate_delay(attempt)
logger.warning(
f"⚠ Verbindungsfehler: {e}. "
f"Retry {attempt + 1}/{self.config.max_retries + 1} "
f"in {delay:.1f}s"
)
if attempt < self.config.max_retries:
self._wait_with_progress(delay, attempt)
continue
else:
break
# Alle Retries erschöpft
self.stats["failed_requests"] += 1
logger.error(f"✗ Alle {self.config.max_retries + 1} Versuche fehlgeschlagen")
raise Exception(
f"Max retries ({self.config.max_retries}) exceeded. "
f"Letzter Fehler: {last_exception}"
)
def get_stats(self) -> Dict[str, Any]:
"""Gibt Statistiken zurück"""
total = self.stats["total_requests"]
success_rate = (
(self.stats["successful_requests"] / total * 100)
if total > 0 else 0
)
return {
**self.stats,
"success_rate": f"{success_rate:.2f}%"
}
============== Beispiel-Nutzung ==============
if __name__ == "__main__":
# Client initialisieren
client = HolySheepRetryClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=RetryConfig(
base_delay=0.5,
max_delay=30.0,
max_retries=5
)
)
# Chat-Completion anfragen
response = client.request_with_retry(
"POST",
"chat/completions",
json={
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": "Erkläre Exponential Backoff"}
],
"max_tokens": 500
}
)
print(f"Antwort: {response['choices'][0]['message']['content']}")
print(f"Statistiken: {client.get_stats()}")Fortgeschrittene Version mit Circuit Breaker Pattern
import time
from enum import Enum
from threading import Lock
from typing import Dict, Any, Optional
from dataclasses import dataclass
class CircuitState(Enum):
"""Zustände des Circuit Breaker"""
CLOSED = "closed" # Normaler Betrieb
OPEN = "open" # Circuit offen, schnelle Fehler
HALF_OPEN = "half_open" # Test-Phase
@dataclass
class CircuitBreakerConfig:
"""Konfiguration für Circuit Breaker"""
failure_threshold: int = 5 # Fehler bis Öffnung
success_threshold: int = 3 # Erfolge zum Schließen
timeout: float = 30.0 # Timeout in Sekunden
half_open_requests: int = 3 # Test-Anfragen im HALF_OPEN
class CircuitBreaker:
"""
Circuit Breaker Pattern Implementierung.
Verhindert Kaskaden-Ausfälle bei kontinuierlichen API-Problemen.
Kombiniert mit Exponential Backoff ergibt sich maximale Resilienz.
"""
def __init__(self, config: Optional[CircuitBreakerConfig] = None):
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_requests_made = 0
self._lock = Lock()
def call(self, func: callable, *args, **kwargs) -> Any:
"""
Führt Funktion mit Circuit Breaker Protection aus.
"""
with self._lock:
# State prüfen
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self._to_half_open()
else:
raise CircuitBreakerOpenError(
"Circuit Breaker ist offen. "
f"Warte noch {self._remaining_timeout():.1f}s"
)
# HALF_OPEN: Nur begrenzte Anfragen zulassen
if self.state == CircuitState.HALF_OPEN:
if self.half_open_requests_made >= self.config.half_open_requests:
raise CircuitBreakerOpenError(
"Circuit Breaker testet noch..."
)
self.half_open_requests_made += 1
# Funktion ausführen
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""Prüft ob Timeout abgelaufen ist"""
if self.last_failure_time is None:
return True
return (time.time() - self.last_failure_time) >= self.config.timeout
def _to_half_open(self):
"""Wechselt zu HALF_OPEN Zustand"""
self.state = CircuitState.HALF_OPEN
self.half_open_requests_made = 0
print("🔄 Circuit Breaker: CLOSED → HALF_OPEN")
def _on_success(self):
"""Behandelt erfolgreichen Aufruf"""
with self._lock:
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
print("✅ Circuit Breaker: HALF_OPEN → CLOSED")
def _on_failure(self):
"""Behandelt fehlgeschlagenen Aufruf"""
with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
print("❌ Circuit Breaker: HALF_OPEN → OPEN (erneuter Fehler)")
elif self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
print(f"❌ Circuit Breaker: CLOSED → OPEN ({self.failure_count} Fehler)")
def _remaining_timeout(self) -> float:
"""Berechnet verbleibende Wartezeit"""
if self.last_failure_time is None:
return 0
elapsed = time.time() - self.last_failure_time
return max(0, self.config.timeout - elapsed)
class CircuitBreakerOpenError(Exception):
"""Exception wenn Circuit Breaker offen ist"""
pass
============== Integrierter Advanced Client ==============
class AdvancedHolySheepClient(HolySheepRetryClient):
"""
Erweiterter Client mit Circuit Breaker Integration.
In meinen Produktionsprojekten mit HolySheep AI hat diese Kombination
die Ausfallzeit um 94% reduziert.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.circuit_breaker = CircuitBreaker()
def request_with_protection(self, method: str, endpoint: str, **kwargs):
"""
Führt Anfrage mit Circuit Breaker + Retry aus.
"""
def _do_request():
return self.request_with_retry(method, endpoint, **kwargs)
return self.circuit_breaker.call(_do_request)
============== Benchmark Beispiel ==============
if __name__ == "__main__":
print("=" * 60)
print("HolySheep AI API Benchmark mit Retry-Strategie")
print("=" * 60)
client = AdvancedHolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=RetryConfig(base_delay=0.1, max_retries=3)
)
# 10 Test-Anfragen durchführen
start_time = time.time()
successful = 0
for i in range(10):
try:
response = client.request_with_protection(
"POST",
"chat/completions",
json={
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": f"Test {i+1}: Kurze Antwort"}
],
"max_tokens": 50
}
)
successful += 1
print(f"✓ Anfrage {i+1}: Erfolgreich (Latenz: {time.time()-start_time:.0f}ms)")
except Exception as e:
print(f"✗ Anfrage {i+1}: Fehlgeschlagen - {e}")
total_time = time.time() - start_time
print(f"\n📊 Ergebnis:")
print(f" Erfolgreich: {successful}/10")
print(f" Gesamtzeit: {total_time:.2f}s")
print(f" Durchschnitt: {total_time/10*1000:.0f}ms pro Anfrage")
print(f" Circuit Breaker Status: {client.circuit_breaker.state.value}")
print(f" Client Statistiken: {client.get_stats()}")Async-Version für High-Throughput-Szenarien
import asyncio
import aiohttp
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
import random
import time
@dataclass
class AsyncRetryConfig:
"""Konfiguration für async Exponential Backoff"""
base_delay: float = 0.1
max_delay: float = 10.0
max_retries: int = 5
jitter: float = 0.15
timeout: float = 30.0
class AsyncHolySheepClient:
"""
Asynchroner API-Client für HolySheep AI mit Exponential Backoff.
Mit HolySheeps <50ms Latenz und dieser async Implementierung
erreiche ich 500+ Anfragen/Sekunde in Batch-Szenarien.
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
config: Optional[AsyncRetryConfig] = None,
semaphore_limit: int = 10
):
self.api_key = api_key
self.base_url = base_url
self.config = config or AsyncRetryConfig()
self.semaphore = asyncio.Semaphore(semaphore_limit)
# Stats
self._stats = {"success": 0, "retry": 0, "fail": 0}
self._stats_lock = asyncio.Lock()
async def _calculate_delay(self, attempt: int) -> float:
"""Berechnet async Wartezeit mit Jitter"""
delay = self.config.base_delay * (2 ** attempt)
delay = min(delay, self.config.max_delay)
jitter = delay * self.config.jitter * random.uniform(-1, 1)
return max(0, delay + jitter)
async def _retry_request(
self,
session: aiohttp.ClientSession,
method: str,
url: str,
**kwargs
) -> Dict[str, Any]:
"""Führt einzelne Anfrage mit Retry durch"""
for attempt in range(self.config.max_retries + 1):
try:
async with session.request(
method, url, **kwargs, timeout=self.config.timeout
) as response:
if response.status == 200:
async with self._stats_lock:
self._stats["success"] += 1
return await response.json()
elif response.status == 429:
async with self._stats_lock:
self._stats["retry"] += 1
delay = await self._calculate_delay(attempt)
print(f"Rate Limit, retry {attempt+1} in {delay:.2f}s")
await asyncio.sleep(delay)
continue
elif response.status >= 500:
async with self._stats_lock:
self._stats["retry"] += 1
delay = await self._calculate_delay(attempt)
print(f"Server Error {response.status}, retry in {delay:.2f}s")
await asyncio.sleep(delay)
continue
else:
text = await response.text()
raise Exception(f"HTTP {response.status}: {text}")
except asyncio.TimeoutError:
async with self._stats_lock:
self._stats["retry"] += 1
delay = await self._calculate_delay(attempt)
print(f"Timeout, retry in {delay:.2f}s")
await asyncio.sleep(delay)
continue
async with self._stats_lock:
self._stats["fail"] += 1
raise Exception(f"Max retries ({self.config.max_retries}) exceeded")
async def request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Thread-sichere Anfrage mit Semaphore"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async with self.semaphore:
timeout = aiohttp.ClientTimeout(total=self.config.timeout)
async with aiohttp.ClientSession(headers=headers) as session:
return await self._retry_request(session, method, url, **kwargs)
async def batch_chat(
self,
prompts: List[str],
model: str = "gpt-4.1",
max_tokens: int = 200
) -> List[Optional[str]]:
"""
Führt mehrere Chat-Anfragen parallel aus.
Beispiel: 100 Prompts in ~20 Sekunden mit HolySheep AI
(bei 50ms Latenz + 5 parallelen Connections)
"""
tasks = []
for prompt in prompts:
task = self.request(
"POST",
"chat/completions",
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": max_tokens
}
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
responses = []
for result in results:
if isinstance(result, Exception):
responses.append(None)
else:
try:
responses.append(
result["choices"][0]["message"]["content"]
)
except (KeyError, IndexError):
responses.append(None)
return responses
def get_stats(self) -> Dict[str, int]:
return self._stats.copy()
============== Async Benchmark ==============
async def run_async_benchmark():
"""Benchmark für async Batch-Verarbeitung"""
client = AsyncHolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=AsyncRetryConfig(
base_delay=0.05,
max_retries=3,
timeout=15.0
),
semaphore_limit=5
)
# 20 Test-Prompts
prompts = [f"Bitte antworte kurz auf: {i+1}" for i in range(20)]
print("🚀 Starte async Batch-Verarbeitung...")
start_time = time.time()
responses = await client.batch_chat(
prompts,
model="gpt-4.1",
max_tokens=30
)
elapsed = time.time() - start_time
print(f"\n📊 Async Benchmark Ergebnis:")
print(f" Prompts: {len(prompts)}")
print(f" Erfolgreich: {sum(1 for r in responses if r)}")
print(f" Gesamtzeit: {elapsed:.2f}s")
print(f" Durchsatz: {len(prompts)/elapsed:.1f} Anfragen/Sekunde")
print(f" Statistiken: {client.get_stats()}")
return responses
if __name__ == "__main__":
# Async Event Loop starten
responses = asyncio.run(run_async_benchmark())
print(f"\nBeispiel-Antwort: {responses[0][:100] if responses[0] else 'None'}...")Häufige Fehler und Lösungen
Fehler 1: Retry-After Header wird ignoriert
# ❌ FALSCH: Retry-After wird nicht berücksichtigt
def bad_retry(url):
for attempt in range(5):
try:
response = requests.get(url)
if response.status_code != 429:
return response.json()
time.sleep(1) # Immer 1 Sekunde warten
except:
pass
return None
✅ RICHTIG: Retry-After Header verwenden
def good_retry(client, url):
for attempt in range(5):
try:
response = client.session.get(url)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Retry-After prüfen
retry_after = response.headers.get("Retry-After")
if retry_after:
# Server gibt explizite Wartezeit vor
wait_time = float(retry_after)
print(f"Server sagt: Warte {wait_time}s")
else:
# Eigenen Exponential Backoff verwenden
wait_time = min(60, 1 * (2 ** attempt))
wait_time += random.uniform(0, 0.5) # Jitter
print(f"Rate Limit, warte {wait_time:.1f}s (Versuch {attempt+1})")
time.sleep(wait_time)
except requests.exceptions.RequestException as e:
wait_time = 1 * (2 ** attempt)
time.sleep(wait_time)
raise Exception("Max retries exceeded")Fehler 2: Fehlender Jitter verursacht Thundering Herd
# ❌ FALSCH: Kein Jitter - alle Clients warten gleichzeitig
def retry_no_jitter(attempt):
delay = 1 * (2 ** attempt) # Immer 1, 2, 4, 8, 16...
print(f"Warte genau {delay}s")
time.sleep(delay)
✅ RICHTIG: Jitter verhindert Synchronisation
def retry_with_jitter(attempt, jitter_factor=0.2):
base_delay = 1 * (2 ** attempt)
# Jitter: Zufällige Variation ±20%
jitter = base_delay * jitter_factor
delay = base_delay + random.uniform(-jitter, jitter)
# Bereich begrenzen
delay = max(0.1, delay)
print(f"Warte {delay:.2f}s (±{jitter:.2f}s Jitter)")
time.sleep(delay)
Noch besser: Exponential Backoff mit Jitter
def exponential_backoff_jitter(attempt, base=1.0, max_delay=60.0, jitter=0.2):
"""
Industry-Standard: Exponential Backoff with Jitter
Varianten:
- Full Jitter: random(0, cap * (2^attempt))
- Equal Jitter: cap * (2^attempt) / 2 + random(0, cap * (2^attempt) / 2)
- Decorrelated: random(base, prev * 3)
"""
# Full Jitter Methode (empfohlen von AWS)
max_cap = min(max_delay, base * (2 ** attempt))
delay = random.uniform(0, max_cap)
print(f"Full Jitter: {delay:.2f}s (max wäre {max_cap:.2f}s)")
return delay
Test verschiedener Jitter-Strategien
print("=== Jitter-Strategien Test ===")
for i in range(5):
print(f"\nVersuch {i}:")
retry_with_jitter(i, 0.2)
exponential_backoff_jitter(i)Fehler 3: Unbegrenzte Retries blockieren das System
# ❌ FALSCH: Unbegrenzte Retries
def infinite_retry(client, url):
attempt = 0
while True:
try:
response = client.get(url)
return response.json()
except:
attempt += 1
delay = min(300, 1 * (2 ** attempt)) # Kann 5+ Minuten erreichen!
print(f"Versuch {attempt}, warte {delay}s...")
time.sleep(delay)
# WIRD NIE ERREICHT!
✅ RICHTIG: Max Retries mit合理em Limit
class RetryManager:
"""Professioneller Retry-Manager mit Max-Limits"""
def __init__(
self,
max_retries: int = 5,
max_total_wait: float = 120.0, # Max 2 Minuten warten
max_delay: float = 30.0 # Max 30s zwischen Versuchen
):
self.max_retries = max_retries
self.max_total_wait = max_total_wait
self.max_delay = max_delay
self.total_wait_time = 0
def should_retry(self, attempt: int) -> bool:
"""Prüft ob Retry sinnvoll ist"""
if attempt >= self.max_retries:
print(f"Max Retries ({self.max_retries}) erreicht")
return False
# Max Gesamtwartezeit prüfen
next_delay = min(self.max_delay, 1 * (2 ** attempt))
if self.total_wait_time + next_delay > self.max_total_wait:
remaining = self.max_total_wait - self.total_wait_time
print(f"Max Wartezeit ({self.max_total_wait}s) erreicht, {remaining:.1f}s verbleibend")
return False
return True
def wait(self, attempt: int):
"""Führt Wartezeit durch und trackt Gesamtwartezeit"""
delay = min(self.max_delay, 1 * (2 ** attempt))
delay = max(0.1, delay)
print(f"Versuch {attempt + 1}: Warte {delay:.1f}s (Gesamt: {self.total_wait_time + delay:.1f}s)")
time.sleep(delay)
self.total_wait_time += delay
def execute_with_retry(self, func, *args, **kwargs):
"""Führt Funktion mit Retry aus"""
for attempt in range(self.max_retries):
if not self.should_retry(attempt):
raise Exception("Retry-Limit erreicht")
try:
return func(*args, **kwargs)
except Exception as e:
print(f"Fehler bei Versuch {attempt + 1}: {e}")
if attempt < self.max_retries - 1:
self.wait(attempt)