Implémentation de la logique de nouvelle tentative avec PydanticAI : Un tutoriel étape par étape
Nous construisons un service de récupération de données fiable qui gère les interruptions réseau comme un pro — car personne ne veut que son application plante à cause d’un simple timeout.
Prérequis
- Python 3.11+
- pip install pydantic-ai
- Compréhension de base des modèles Pydantic
- Connaissance des bibliothèques de requêtes réseau (comme requests ou httpx)
Étape 1 : Configuration de votre environnement
Avant d’implémenter la logique de nouvelle tentative, nous devons configurer notre environnement de développement. Cela impliquera d’installer le package PydanticAI et de s’assurer que nous avons toutes les bibliothèques nécessaires.
# Préparez votre environnement Python
# Installez les bibliothèques pydantic-ai et httpx
!pip install pydantic-ai httpx
Cette commande installe PydanticAI ainsi que httpx, qui est crucial pour effectuer des requêtes HTTP. J’ai choisi httpx plutôt que requests en raison de son support natif pour les opérations asynchrones, ce qui est un avantage de poids pour les performances. Maintenant, si vous venez de l’univers des requêtes, soyez prêt à une légère courbe d’apprentissage, mais croyez-moi, cela en vaut la peine.
Étape 2 : Création d’un modèle Pydantic de base
Avec l’environnement prêt, créons notre premier modèle Pydantic. Un modèle dans Pydantic définit le schéma pour vos données – c’est comme un plan. Nous allons en créer un qui gère une réponse API simulée.
from pydantic import BaseModel
class ApiResponseModel(BaseModel):
status: str
data: dict
Ce modèle a deux champs : status et data. Le champ status indique si la requête a réussi, et data contient le contenu de la réponse. La partie intelligente ici est que Pydantic va valider la structure de la réponse API par rapport à ce modèle.
Étape 3 : Implémentation de la logique de requête HTTP
Maintenant, créons une fonction pour effectuer une requête HTTP. Nous utiliserons Pydantic pour valider la réponse. Voici le bon côté : nous allons ajouter une logique de nouvelle tentative, donc si notre requête échoue, nous pourrons réessayer au lieu de faire une crise.
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())
Ce code récupère des données depuis l’URL spécifiée. S’il échoue, nous devons implémenter la logique de nouvelle tentative. Personne ne veut gérer des erreurs HTTP en production, n’est-ce pas ?
Étape 4 : Ajout de la logique de nouvelle tentative
Il est temps d’ajouter cette logique de nouvelle tentative si importante. Nous allons mettre en œuvre un mécanisme simple pour réessayer la requête un nombre spécifié de fois avant d’abandonner.
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"Tentative {i + 1} échouée : {e}. Nouvelle tentative...")
time.sleep(delay) # Attendre avant de réessayer
except Exception as e:
print(f"Erreur inattendue : {e}. Nouvelle tentative...")
time.sleep(delay)
raise Exception(f"Echec de la récupération des données après {retries} tentatives.")
Dans cette fonction, nous prenons un paramètre optionnel retries pour nous permettre de spécifier combien de fois tenter la requête avant d’abandonner. Nous avons aussi un paramètre delay, donc si notre requête échoue, nous faisons une pause de quelques secondes au lieu de bombarder le serveur avec des requêtes.
Les pièges
Jusqu’ici, tout va bien, mais voici quelques pièges dans lesquels je suis tombé maintes fois et qui peuvent vous faire trébucher en production :
- Exponential Backoff : Il est crucial d’éviter de surcharger le serveur après un échec. Assurez-vous d’augmenter le délai à chaque nouvelle tentative. Des délais statiques simples ne suffiront pas lorsque le système commence à surcharger le serveur.
- Gestion des différentes exceptions : Vous voudrez peut-être ne réessayer que sur des problèmes temporaires. Toutes les exceptions ne méritent pas une nouvelle tentative, alors affinez votre gestion des exceptions pour capturer des erreurs spécifiques.
- Journalisation des échecs : N’oubliez pas de consigner des détails sur les échecs. C’est un précieux témoin pour diagnostiquer des problèmes plus tard.
- Gestionnaires de contexte asynchrones : Si vous mélangez du code synchrone et asynchrone, faites attention aux gestionnaires de contexte. Ils ne fonctionnent pas de la même manière, et cela peut entraîner des bogues désagréables.
- Latence réseau : Soyez conscient que la latence réseau peut affecter considérablement vos nouvelles tentatives. Testez dans différentes conditions pour structurer votre logique autour de scénarios réalistes.
Exemple de code complet
Voici une version complète de ce que nous avons construit, entièrement opérationnelle. Cela facilitera la copie et l’utilisation immédiate. Remplacez simplement YOUR_API_URL_HERE par votre point de terminaison cible.
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"Tentative {i + 1} échouée : {e}. Nouvelle tentative...")
time.sleep(delay)
except Exception as e:
print(f"Erreur inattendue : {e}. Nouvelle tentative...")
time.sleep(delay)
raise Exception(f"Echec de la récupération des données après {retries} tentatives.")
# Exemple d'utilisation
# Assurez-vous que cela soit exécuté dans un contexte asynchrone, comme asyncio.run()
url = "YOUR_API_URL_HERE"
data = await fetch_with_retries(url)
print(data)
Quelle est la suite ?
Votre prochaine étape ? Expérimentez en ajoutant des journaux personnalisés et des métriques à votre logique de nouvelle tentative. Vous voudrez surveiller à quelle fréquence les requêtes échouent, le temps pris pour les tentatives de nouvelle tentative, etc. Avoir ces données peut améliorer considérablement la fiabilité de votre application.
FAQ
Q : Pourquoi utiliser PydanticAI plutôt que les options traditionnelles ?
R : PydanticAI permet une gestion structurée des données et une validation automatique de vos réponses API, quelque chose de désagréable à gérer manuellement. Avec 15,593 étoiles sur GitHub, PydanticAI a prouvé son succès, ce qui peut signifier un meilleur support communautaire et des ressources.
Q : Que faire si mon API nécessite une authentification ?
R : Vous devrez étendre le client HTTP pour inclure des en-têtes d’authentification. Cela peut être fait en modifiant l’initialisation de httpx.AsyncClient() pour inclure un paramètre auth si votre API supporte l’authentification basique, ou en utilisant des jetons OAuth selon les besoins.
Q : Existe-t-il une alternative à la logique de nouvelle tentative dans les applications en temps réel ?
R : Bien que la logique de nouvelle tentative soit essentielle, envisagez d’implémenter des modèles de disjoncteur comme couche supplémentaire de résilience. Cela permet à votre application de réagir plus gracieusement aux échecs et peut réduire la charge du serveur lors des pannes.
Recommandations pour différents profils de développeurs
Si vous êtes nouveau dans cet univers de récupération de données :
- Nouveau développeur : Concentrez-vous sur la maîtrise des bases, familiarisez-vous avec l’envoi de requêtes HTTP, et structurez vos modèles selon des API réelles.
- Développeur intermédiaire : Regardez des mécanismes de nouvelle tentative avancés et des stratégies d’optimisation des performances. Envisagez de créer une méthode de nouvelle tentative personnalisée pour plus de flexibilité.
- Développeur senior : Prenez le temps de peaufiner la gestion des erreurs et les modèles d’observateur qui peuvent fournir de meilleures informations sur le comportement de votre application sous charge. Les métriques et la journalisation sont vos amis !
Données au 20 mars 2026. Sources : pydantic/pydantic-ai, Pydantic AI Retries, Ajoutez un comportement personnalisé de nouvelle tentative des requêtes.
Articles connexes
- Midjourney News Octobre 2025 : La prochaine avancée de l’art IA révélée !
- Mon entreprise locale LLMs avec RAG
- Mon moteur de requête LlamaIndex gère des données complexes
🕒 Published: