Implementierung der Wiederholungslogik mit PydanticAI: Ein Schritt-für-Schritt-Tutorial
Wir bauen einen zuverlässigen Datenerfassungsdienst auf, der Netzwerkunterbrechungen wie ein Profi verwaltet – denn niemand möchte, dass seine Anwendung wegen eines einfachen Timeouts abstürzt.
Voraussetzungen
- Python 3.11+
- pip install pydantic-ai
- Grundverständnis der Pydantic-Modelle
- Kenntnisse über Netzwerk-Anfragebibliotheken (wie requests oder httpx)
Schritt 1: Konfiguration Ihrer Umgebung
Bevor wir die Wiederholungslogik implementieren, müssen wir unsere Entwicklungsumgebung einrichten. Dazu gehört die Installation des PydanticAI-Pakets und die Sicherstellung, dass wir alle notwendigen Bibliotheken haben.
# Richten Sie Ihre Python-Umgebung ein
# Installieren Sie die Bibliotheken pydantic-ai und httpx
!pip install pydantic-ai httpx
Dieser Befehl installiert PydanticAI sowie httpx, das entscheidend für HTTP-Anfragen ist. Ich habe mich für httpx anstelle von requests entschieden, weil es nativen Support für asynchrone Operationen bietet, was ein gewaltiger Vorteil für die Leistung ist. Wenn Sie aus der Welt der Anfragen kommen, seien Sie auf eine kleine Lernkurve vorbereitet, aber glauben Sie mir, es lohnt sich.
Schritt 2: Erstellung eines einfachen Pydantic-Modells
Mit einer bereitstehenden Umgebung lassen Sie uns unser erstes Pydantic-Modell erstellen. 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 verwaltet.
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 den Inhalt der Antwort. Der clevere Teil hier ist, dass Pydantic die Struktur der API-Antwort gegen dieses Modell validiert.
Schritt 3: Implementierung der HTTP-Anfrage-Logik
Jetzt erstellen wir eine Funktion, um eine HTTP-Anfrage durchzuführen. Wir werden Pydantic verwenden, um die Antwort zu validieren. Der gute Teil: Wir werden eine Wiederholungslogik hinzufügen, sodass wir, wenn unsere Anfrage fehlschlägt, sie erneut versuchen können, anstatt in Panik zu geraten.
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 dies fehlschlägt, müssen wir die Wiederholungslogik implementieren. Niemand möchte HTTP-Fehler in der Produktion behandeln, oder?
Schritt 4: Hinzufügen der Wiederholungslogik
Es ist an der Zeit, diese wichtige Wiederholungslogik hinzuzufügen. Wir werden einen einfachen Mechanismus implementieren, um die Anfrage eine festgelegte Anzahl von Versuchen zu wiederholen, bevor wir 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}. Neuer Versuch...")
time.sleep(delay) # Warten, bevor wir es erneut versuchen
except Exception as e:
print(f"Unerwarteter Fehler: {e}. Neuer Versuch...")
time.sleep(delay)
raise Exception(f"Fehler beim Abrufen der Daten nach {retries} Versuchen.")
In dieser Funktion nehmen wir einen optionalen Parameter retries, um festzulegen, wie oft wir die Anfrage vor dem Aufgeben versuchen wollen. Wir haben auch einen Parameter delay, sodass wir, wenn unsere Anfrage fehlschlägt, ein paar Sekunden pausieren, anstatt den Server mit Anfragen zu bombardieren.
Die Fallstricke
Bisher läuft alles gut, aber hier sind einige Fallstricke, in die ich schon oft geraten bin und die Ihnen in der Produktion Probleme bereiten können:
- Exponential Backoff: Es ist entscheidend, zu vermeiden, dass der Server nach einem Fehler überlastet wird. Stellen Sie sicher, dass Sie die Verzögerung bei jedem neuen Versuch erhöhen. Einfache statische Verzögerungen reichen nicht aus, wenn das System beginnt, den Server überlasten.
- Umgang mit verschiedenen Ausnahmen: Sie möchten vielleicht nur bei vorübergehenden Problemen wiederholen. Nicht alle Ausnahmen verdienen einen neuen Versuch, also verfeinern Sie Ihr Ausnahmehandling, um spezifische Fehler zu erfassen.
- Protokollierung der Fehler: Vergessen Sie nicht, Details zu den Fehlern aufzuzeichnen. Dies ist ein wertvoller Zeuge, um später Probleme zu diagnostizieren.
- Asynchrone Kontext-Manager: Wenn Sie synchronen und asynchronen Code mischen, seien Sie vorsichtig mit den Kontext-Manager. Sie funktionieren nicht auf die gleiche Weise, und das kann zu unangenehmen Bugs führen.
- Netzwerklatenz: Seien Sie sich bewusst, dass die Netzwerklatenz Ihre Wiederholungsversuche erheblich beeinträchtigen 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 aufgebaut haben, die vollständig funktionsfähig ist. Dies erleichtert das Kopieren und die sofortige Nutzung. Ersetzen Sie einfach YOUR_API_URL_HERE durch Ihren Zielendpunkt.
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}. Neuer Versuch...")
time.sleep(delay)
except Exception as e:
print(f"Unerwarteter Fehler: {e}. Neuer Versuch...")
time.sleep(delay)
raise Exception(f"Fehler beim Abrufen der Daten nach {retries} Versuchen.")
# Beispiel für die Verwendung
# Stellen Sie sicher, dass dies in einem asynchronen Kontext ausgeführt wird, 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, indem Sie benutzerdefinierte Protokolle und Metriken in Ihre Wiederholungslogik integrieren. Sie möchten überwachen, wie oft Anfragen fehlschlagen, die Zeit für Wiederholungsversuche usw. Diese Daten zu haben kann die Zuverlässigkeit Ihrer Anwendung erheblich verbessern.
FAQ
F: Warum PydanticAI anstelle herkömmlicher Optionen verwenden?
ANTWORT: PydanticAI ermöglicht ein strukturiertes Datenmanagement und die automatische Validierung Ihrer API-Antworten, was es unangenehm macht, manuell zu verwalten. Mit 15.593 Sternen auf GitHub hat PydanticAI seinen Erfolg bewiesen, was besseren Community-Support und Ressourcen bedeuten kann.
F: Was, wenn meine API eine Authentifizierung benötigt?
ANTWORT: Sie müssen den HTTP-Client erweitern, um Authentifizierungsheader einzuschließen. Dies kann erreicht werden, indem Sie die Initialisierung von httpx.AsyncClient() ändern, um einen auth-Parameter hinzuzufügen, wenn Ihre API die grundlegende Authentifizierung unterstützt, oder indem Sie bei Bedarf OAuth-Token verwenden.
F: Gibt es eine Alternative zur Wiederholungslogik in Echtzeitanwendungen?
ANTWORT: Obwohl die Wiederholungslogik unerlässlich ist, ziehen Sie in Betracht, Ausfallmuster als zusätzliche Resilienzschicht zu implementieren. Dadurch kann Ihre Anwendung eleganter auf Fehler reagieren und die Serverlast während Ausfällen reduzieren.
Empfehlungen für verschiedene Entwicklerprofile
Wenn Sie neu in der Welt der Datenerfassung sind:
- Neuer Entwickler: Konzentrieren Sie sich darauf, die Grundlagen zu meistern, machen Sie sich mit dem Senden von HTTP-Anfragen vertraut und strukturieren Sie Ihre Modelle gemäß echten APIs.
- Fortgeschrittener Entwickler: Schauen Sie sich fortgeschrittene Wiederholungsmechanismen und Leistungsoptimierungsstrategien an. Ziehen Sie in Betracht, eine benutzerdefinierte Wiederholungsmethode für mehr Flexibilität zu erstellen.
- Senior Entwickler: Nehmen Sie sich Zeit, um das Fehlerhandling und die Beobachtungsmodelle zu verfeinern, die bessere Einblicke in das Verhalten Ihrer Anwendung unter Last liefern können. Metriken und Protokollierung sind Ihre Freunde!
Daten vom 20. März 2026. Quellen: pydantic/pydantic-ai, Pydantic AI Retries, Fügen Sie benutzerverhalten für Wiederholungsanfragen hinzu.
Verwandte Artikel
- Midjourney News Oktober 2025: Der nächste Sprung in der KI-Kunst enthüllt!
- Mein lokales Unternehmen LLMs mit RAG
- Mein LlamaIndex-Abfrage-Engine verarbeitet komplexe Daten
🕒 Published: