Implémentation de la logique de répétition 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 — parce que personne ne veut que son application plante à cause d’un simple délai d’attente.
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 de mettre en œuvre la logique de répétition, nous devons configurer notre environnement de développement. Cela implique 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 essentiel pour faire des requêtes HTTP. J’ai choisi httpx plutôt que requests en raison de sa prise en charge native des opérations asynchrones, ce qui est un grand avantage pour les performances. Maintenant, si vous venez de l’univers de requests, préparez-vous à une légère courbe d’apprentissage, mais croyez-moi, ça 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 de vos données – c’est comme un blueprint. Nous allons en créer un qui gère une réponse d’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 la charge utile de la réponse. La partie intelligente ici est que Pydantic va valider la structure de la réponse de l’API par rapport à ce modèle.
Étape 3 : Mise en œuvre de la logique de requête HTTP
Maintenant, créons une fonction pour faire une requête HTTP. Nous allons utiliser Pydantic pour valider la réponse. Voici le bonus : nous allons ajouter une logique de répétition, donc si notre requête échoue, nous pouvons essayer à nouveau au lieu de faire une crise de nerfs.
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. Si cela échoue, nous devons mettre en œuvre la logique de répétition. Personne ne veut traiter des erreurs HTTP en production, n’est-ce pas ?
Étape 4 : Ajout de la logique de répétition
Il est temps d’ajouter cette logique de répétition 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 enfin.
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"Échec 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 également 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
Jusque-là, tout va bien, mais voici quelques pièges dans lesquels je suis tombé maintes fois et qui peuvent vous gêner en production :
- Retard exponentiel : Il est crucial d’éviter de surcharger le serveur après un échec. Assurez-vous d’augmenter le délai chaque fois que vous réessayez. Des délais statiques simples ne suffiront pas lorsque le système commence à surcharger le serveur.
- Gestion des différentes exceptions : Vous pourriez vouloir réessayer uniquement en cas de problèmes transitoires. Toutes les exceptions ne justifient pas un nouveau essai, donc affinez votre gestion des exceptions pour attraper des erreurs spécifiques.
- Journalisation des échecs : N’oubliez pas de consigner les détails des échecs. C’est un sauveur pour diagnostiquer des problèmes plus tard.
- Gestionnaires de contexte asynchrones : Si vous mélangez du code sync et async, faites attention aux gestionnaires de contexte. Ils ne fonctionnent pas de la même manière, et cela peut conduire à des bogues désagréables.
- Latence réseau : Soyez conscient que la latence réseau peut affecter considérablement vos répétitions. 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, totalement opérationnelle. Cela vous facilitera la tâche pour copier et utiliser immédiatement. Il suffit de remplacer 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"Échec 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 async, comme asyncio.run()
url = "YOUR_API_URL_HERE"
data = await fetch_with_retries(url)
print(data)
Quelle est la suite ?
Votre prochain mouvement ? Expérimentez en ajoutant une journalisation personnalisée et des métriques à votre logique de répétition. Vous voudrez surveiller à quelle fréquence les requêtes échouent, le temps pris pour les tentatives de répétition, etc. Avoir ces données peut considérablement améliorer la fiabilité de votre application.
FAQ
Q : Pourquoi utiliser PydanticAI plutôt que des options traditionnelles ?
A : PydanticAI permet une gestion structurée des données et une validation automatique de vos réponses API, ce qui est fastidieux à gérer manuellement. Avec 15 593 étoiles sur GitHub, PydanticAI a prouvé sa popularité, ce qui peut signifier un meilleur support communautaire et des ressources.
Q : Que faire si mon API nécessite une authentification ?
A : 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 prend en charge l’authentification basique, ou en utilisant des jetons OAuth selon les besoins.
Q : Existe-t-il une alternative à la logique de répétition dans des applications en temps réel ?
A : Bien que la logique de répétition soit essentielle, envisagez de mettre en œuvre des modèles de disjoncteurs comme couche supplémentaire de résilience. Cela permet à votre application de répondre aux échecs de manière plus élégante et peut réduire la charge du serveur pendant les pannes.
Recommandations pour différents profils de développeur
Si vous êtes nouveau dans ce monde de la récupération de données :
- Nouveau développeur : Concentrez-vous sur la maîtrise des bases, familiarisez-vous avec la réalisation de requêtes HTTP et modèlez vos modèles selon de vraies API.
- Développeur intermédiaire : Explorez des mécanismes de répétition avancés et des stratégies d’optimisation des performances. Envisagez de construire une méthode de répétition 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 alliées !
Données à partir du 20 mars 2026. Sources : pydantic/pydantic-ai, Pydantic AI Retries, Ajouter un comportement de répétition de requête personnalisé.
Articles connexes
- Actualités Midjourney octobre 2025 : le prochain saut de l’art AI révélé !
- Mon entreprise locale avec des LLMs localisés et RAG
- Mon moteur de requêtes LlamaIndex gère des données complexes
🕒 Published: