Implementierung von Retry-Logik mit PydanticAI: Ein Schritt-für-Schritt-Tutorial
Wir bauen einen zuverlässigen Datenabrufdienst, der Netzwerkunterbrechungen wie ein Profi bewältigt – denn niemand möchte, dass seine App wegen eines einfachen Timeouts abstürzt.
Voraussetzungen
- Python 3.11+
- pip install pydantic-ai
- Grundverständnis von Pydantic-Modellen
- Kenntnisse über Netzwerk-Anforderungsbibliotheken (wie requests oder httpx)
Schritt 1: Einrichten Ihrer Umgebung
Bevor wir die Retry-Logik implementieren, müssen wir unsere Entwicklungsumgebung einrichten. Dazu gehört die Installation des PydanticAI-Pakets und die Sicherstellung, dass wir alle notwendigen Bibliotheken haben.
# Bereiten Sie Ihre Python-Umgebung vor
# Installieren Sie die Bibliotheken pydantic-ai und httpx
!pip install pydantic-ai httpx
Dieser Befehl installiert PydanticAI zusammen mit httpx, das für HTTP-Anfragen entscheidend ist. Ich habe mich für httpx anstelle von requests entschieden, da es native Unterstützung für asynchrone Vorgänge bietet, was für die Leistung ein großer Vorteil ist. Wenn Sie von einem requests-Hintergrund kommen, seien Sie auf eine kleine Lernkurve vorbereitet, aber vertrauen Sie mir, es lohnt sich.
Schritt 2: Erstellen eines einfachen Pydantic-Modells
Mit der eingerichteten Umgebung erstellen wir unser erstes Pydantic-Modell. Ein Modell in Pydantic definiert das Schema für Ihre Daten – es ist wie ein Bauplan. Wir werden eines erstellen, das eine simulierte API-Antwort verarbeitet.
from pydantic import BaseModel
class ApiResponseModel(BaseModel):
status: str
data: dict
Dieses Modell hat zwei Felder: status und data. Das Feld status gibt an, ob die Anfrage erfolgreich war, und data enthält die Antwortdaten. Der clevere Teil hier ist, dass Pydantic die Struktur der API-Antwort gegen dieses Modell validieren wird.
Schritt 3: Implementierung der HTTP-Anforderungslogik
Jetzt erstellen wir eine Funktion, um eine HTTP-Anfrage zu machen. Wir werden Pydantic verwenden, um die Antwort zu validieren. Hier ist der Knackpunkt: wir fügen die Retry-Logik hinzu, sodass, wenn unsere Anfrage fehlschlägt, wir es erneut versuchen können, anstatt einen Wutanfall zu bekommen.
import httpx
import time
async def fetch_data(url: str) -> ApiResponseModel:
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status()
return ApiResponseModel(**response.json())
Dieser Code ruft Daten von der angegebenen URL ab. Wenn es fehlschlägt, müssen wir die Retry-Logik implementieren. Niemand möchte in der Produktion mit HTTP-Fehlern umgehen, oder?
Schritt 4: Hinzufügen der Retry-Logik
Es ist an der Zeit, diese wichtige Retry-Logik hinzuzufügen. Wir werden einen einfachen Mechanismus implementieren, um die Anfrage eine bestimmte Anzahl von Versuchen zu wiederholen, bevor wir endgültig aufgeben.
async def fetch_with_retries(url: str, retries: int = 3, delay: int = 2) -> ApiResponseModel:
for i in range(retries):
try:
return await fetch_data(url)
except httpx.HTTPStatusError as e:
print(f"Versuch {i + 1} fehlgeschlagen: {e}. Erneut versuchen...")
time.sleep(delay) # Warten vor dem erneuten Versuch
except Exception as e:
print(f"Unerwarteter Fehler: {e}. Erneut versuchen...")
time.sleep(delay)
raise Exception(f"Fehler beim Abrufen der Daten nach {retries} Versuchen.")
In dieser Funktion nehmen wir einen optionalen Parameter retries, um anzugeben, wie oft wir die Anfrage versuchen möchten, bevor wir aufgeben. Wir haben auch einen Parameter delay, sodass, wenn unsere Anfrage fehlschlägt, wir ein paar Sekunden warten, anstatt den Server mit Anfragen zu bombardieren.
Die Stolpersteine
Bisher ist alles gut, aber hier sind einige Fallstricke, in die ich immer wieder gestolpert bin und die Sie in der Produktion aufhalten können:
- Exponential Backoff: Es ist entscheidend, um zu vermeiden, dass der Server nach einem Fehler überlastet wird. Stellen Sie sicher, dass Sie die Verzögerung jedes Mal erhöhen, wenn Sie erneut versuchen. Einfache statische Verzögerungen sind nicht ausreichend, wenn das System den Server überlastet.
- Umgang mit verschiedenen Ausnahmen: Möglicherweise möchten Sie nur bei vorübergehenden Problemen einen erneuten Versuch unternehmen. Nicht alle Ausnahmen rechtfertigen einen erneuten Versuch, also verfeinern Sie Ihre Fehlerbehandlung, um spezifische Fehler abzufangen.
- Fehlerprotokollierung: Vergessen Sie nicht, Details zu den Fehlern zu protokollieren. Es ist eine Lebensretter für die spätere Diagnose von Problemen.
- Asynchrone Kontextmanager: Wenn Sie synchronen und asynchronen Code mischen, achten Sie auf die Kontextmanager. Sie funktionieren nicht auf die gleiche Weise und können zu unangenehmen Bugs führen.
- Netzwerklatenz: Seien Sie sich bewusst, dass die Netzwerklatenz Ihre Versuche erheblich beeinflussen kann. Testen Sie unter verschiedenen Bedingungen, um Ihre Logik um realistische Szenarien herum zu strukturieren.
Vollständiges Codebeispiel
Hier ist eine vollständige Version dessen, was wir gebaut haben, voll funktionsfähig. Dies macht es Ihnen leicht, sie sofort zu übernehmen und zu verwenden. Ersetzen Sie einfach YOUR_API_URL_HERE durch Ihre Ziel-URL.
import httpx
import time
from pydantic import BaseModel
class ApiResponseModel(BaseModel):
status: str
data: dict
async def fetch_data(url: str) -> ApiResponseModel:
async with httpx.AsyncClient() as client:
response = await client.get(url)
response.raise_for_status()
return ApiResponseModel(**response.json())
async def fetch_with_retries(url: str, retries: int = 3, delay: int = 2) -> ApiResponseModel:
for i in range(retries):
try:
return await fetch_data(url)
except httpx.HTTPStatusError as e:
print(f"Versuch {i + 1} fehlgeschlagen: {e}. Erneut versuchen...")
time.sleep(delay)
except Exception as e:
print(f"Unerwarteter Fehler: {e}. Erneut versuchen...")
time.sleep(delay)
raise Exception(f"Fehler beim Abrufen der Daten nach {retries} Versuchen.")
# Beispielverwendung
# Stellen Sie sicher, dass dies in einem asynchronen Kontext ausgeführt wird, wie z.B. asyncio.run()
url = "YOUR_API_URL_HERE"
data = await fetch_with_retries(url)
print(data)
Was kommt als Nächstes?
Ihr nächster Schritt? Experimentieren Sie mit dem Hinzufügen von benutzerdefinierter Protokollierung und Metriken zu Ihrer Retry-Logik. Sie möchten überwachen, wie oft Anfragen fehlschlagen, die Zeit, die für Wiederholungsversuche benötigt wird, usw. Diese Daten zu haben, kann die Zuverlässigkeit Ihrer Anwendung erheblich verbessern.
FAQ
F: Warum PydanticAI anstelle traditioneller Optionen verwenden?
A: PydanticAI ermöglicht eine strukturierte Datenverarbeitung und automatische Validierung Ihrer API-Antworten, was manuell schwer zu verwalten ist. Mit 15.593 Sternen auf GitHub hat sich PydanticAI als beliebt erwiesen, was bessere Unterstützung durch die Community und Ressourcen bedeuten kann.
F: Was, wenn meine API Authentifizierung benötigt?
A: Sie müssen den HTTP-Client erweitern, um Authentifizierungsheader einzuschließen. Dies könnte durch die Modifizierung der httpx.AsyncClient() Initialisierung geschehen, um einen auth Parameter einzuschließen, wenn Ihre API die Basis-Authentifizierung unterstützt, oder durch die Verwendung von OAuth-Tokens nach Bedarf.
F: Gibt es eine Alternative zur Retry-Logik in Echtzeitanwendungen?
A: Während die Retry-Logik wichtig ist, sollten Sie in Betracht ziehen, Muster des Stromkreisbrechers als zusätzliche Schutzschicht zu implementieren. Dies ermöglicht Ihrer App, auf Fehler eleganter zu reagieren und kann die Serverlast während Ausfällen reduzieren.
Empfehlungen für verschiedene Entwicklerpersönlichkeiten
Wenn Sie neu in dieser Welt des Datenfetching sind:
- Neuer Entwickler: Konzentrieren Sie sich darauf, die Grundlagen zu beherrschen, werden Sie sicher im Umgang mit HTTP-Anfragen und orientieren Sie Ihre Modelle an echten APIs.
- Fortgeschrittener Entwickler: Schauen Sie sich fortgeschrittene Retry-Mechanismen und Strategien zur Leistungsoptimierung an. Erwägen Sie, eine benutzerdefinierte Retry-Methode für mehr Flexibilität zu entwickeln.
- Senior Entwickler: Verbringen Sie Zeit damit, die Fehlerbehandlung und Beobachtungmuster zu verfeinern, die Ihnen bessere Einblicke in das Verhalten Ihrer Anwendung unter Last geben können. Metriken und Protokollierung sind Ihre besten Freunde!
Daten von 20. März 2026. Quellen: pydantic/pydantic-ai, Pydantic AI Retries, Add Custom Request Retry Behavior.
Verwandte Artikel
- Midjourney-News Oktober 2025: Der nächste Sprung der KI-Kunst enthüllt!
- Mein kleines Unternehmen lokalisierte LLMs mit RAG
- Mein LlamaIndex-Abfrage-Engine verarbeitet komplexe Daten
🕒 Published: