Implementazione della logica di nuova tentativo con PydanticAI: Un tutorial passo dopo passo
Stiamo costruendo un servizio di recupero dati affidabile che gestisce le interruzioni di rete come un professionista — perché nessuno vuole che la propria applicazione si blocchi a causa di un semplice timeout.
Prerequisiti
- Python 3.11+
- pip install pydantic-ai
- Comprensione di base dei modelli Pydantic
- Conoscenza delle librerie di richieste di rete (come requests o httpx)
Passo 1: Configurazione del tuo ambiente
Prima di implementare la logica di nuova tentativo, dobbiamo configurare il nostro ambiente di sviluppo. Questo comporterà l’installazione del pacchetto PydanticAI e assicurarsi di avere tutte le librerie necessarie.
# Prepara il tuo ambiente Python
# Installa le librerie pydantic-ai e httpx
!pip install pydantic-ai httpx
Questo comando installa PydanticAI così come httpx, che è cruciale per effettuare richieste HTTP. Ho scelto httpx piuttosto che requests per il suo supporto nativo delle operazioni asincrone, che è un grande vantaggio per le prestazioni. Ora, se provieni dal mondo delle richieste, preparati a una leggera curva di apprendimento, ma credimi, ne vale la pena.
Passo 2: Creazione di un modello Pydantic di base
Con l’ambiente pronto, creiamo il nostro primo modello Pydantic. Un modello in Pydantic definisce lo schema per i tuoi dati – è come un piano. Ne creeremo uno che gestisce una risposta API simulata.
from pydantic import BaseModel
class ApiResponseModel(BaseModel):
status: str
data: dict
Questo modello ha due campi: status e data. Il campo status indica se la richiesta ha avuto successo, e data contiene il contenuto della risposta. La parte intelligente qui è che Pydantic convaliderà la struttura della risposta API rispetto a questo modello.
Passo 3: Implementazione della logica di richiesta HTTP
Ora, creiamo una funzione per effettuare una richiesta HTTP. Utilizzeremo Pydantic per convalidare la risposta. Ecco il lato positivo: aggiungeremo una logica di nuova tentativo, quindi se la nostra richiesta fallisce, possiamo riprovare invece di andare nel panico.
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())
Questo codice recupera dati dall’URL specificato. Se fallisce, dobbiamo implementare la logica di nuova tentativo. Nessuno vuole gestire errori HTTP in produzione, giusto?
Passo 4: Aggiunta della logica di nuova tentativo
È tempo di aggiungere questa importantissima logica di nuova tentativo. Implementeremo un meccanismo semplice per ripetere la richiesta un numero specificato di volte prima di rinunciare.
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"Tentativo {i + 1} fallito: {e}. Nuovo tentativo...")
time.sleep(delay) # Attendere prima di riprovare
except Exception as e:
print(f"Errore inatteso: {e}. Nuovo tentativo...")
time.sleep(delay)
raise Exception(f"Impossibile recuperare i dati dopo {retries} tentativi.")
In questa funzione, prendiamo un parametro opzionale retries per permetterci di specificare quante volte tentare la richiesta prima di rinunciare. Abbiamo anche un parametro delay, quindi se la nostra richiesta fallisce, facciamo una pausa di alcuni secondi invece di bombardare il server con richieste.
I tranelli
Finora tutto bene, ma ecco alcuni tranelli in cui sono inciampato molte volte e che possono farti cadere in produzione:
- Exponential Backoff: È cruciale evitare di sovraccaricare il server dopo un fallimento. Assicurati di aumentare il ritardo a ogni nuovo tentativo. Ritardi statici semplici non sono sufficienti quando il sistema inizia a sovraccaricare il server.
- Gestione delle diverse eccezioni: Potresti voler riprovare solo su problemi temporanei. Non tutte le eccezioni meritano un nuovo tentativo, quindi affina la tua gestione delle eccezioni per catturare errori specifici.
- Registrazione dei fallimenti: Non dimenticare di registrare dettagli sui fallimenti. È un testimone prezioso per diagnosticare problemi in seguito.
- Gestori di contesto asincroni: Se mescoli codice sincrono e asincrono, fai attenzione ai gestori di contesto. Non funzionano allo stesso modo, e questo può portare a bug indesiderati.
- Latente di rete: Sii consapevole che la latenza di rete può influenzare significativamente i tuoi nuovi tentativi. Testa in diverse condizioni per strutturare la tua logica attorno a scenari realistici.
Esempio di codice completo
Qui c’è una versione completa di ciò che abbiamo costruito, completamente funzionante. Questo faciliterà la copia e l’uso immediato. Sostituisci semplicemente YOUR_API_URL_HERE con il tuo endpoint di destinazione.
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"Tentativo {i + 1} fallito: {e}. Nuovo tentativo...")
time.sleep(delay)
except Exception as e:
print(f"Errore inatteso: {e}. Nuovo tentativo...")
time.sleep(delay)
raise Exception(f"Impossibile recuperare i dati dopo {retries} tentativi.")
# Esempio di utilizzo
# Assicurati che venga eseguito in un contesto asincrono, come asyncio.run()
url = "YOUR_API_URL_HERE"
data = await fetch_with_retries(url)
print(data)
Qual è il prossimo passo?
Il tuo prossimo passo? Sperimenta aggiungendo registri personalizzati e metriche alla tua logica di nuova tentativo. Vorrai monitorare con quale frequenza le richieste falliscono, il tempo impiegato per i tentativi di nuova tentativo, ecc. Avere questi dati può migliorare notevolmente l’affidabilità della tua applicazione.
FAQ
Q: Perché usare PydanticAI piuttosto che le opzioni tradizionali?
R: PydanticAI consente una gestione strutturata dei dati e una convalida automatica delle tue risposte API, qualcosa di spiacevole da gestire manualmente. Con 15.593 stelline su GitHub, PydanticAI ha dimostrato il suo successo, il che può significare un miglior supporto comunitario e risorse.
Q: Cosa fare se la mia API richiede autenticazione?
R: Dovrai estendere il client HTTP per includere intestazioni di autenticazione. Questo può essere fatto modificando l’inizializzazione di httpx.AsyncClient() per includere un parametro auth se la tua API supporta l’autenticazione di base, o utilizzando token OAuth secondo necessità.
Q: Esiste un’alternativa alla logica di nuova tentativo nelle applicazioni in tempo reale?
R: Sebbene la logica di nuova tentativo sia essenziale, considera di implementare modelli di interruttore come ulteriore strato di resilienza. Questo consente alla tua applicazione di reagire più agevolmente ai fallimenti e può ridurre il carico del server durante i crash.
Raccomandazioni per diversi profili di sviluppatori
Se sei nuovo in questo mondo del recupero dati:
- Sviluppatore principiante: Concentrati sul padroneggiare le basi, familiarizza con l’invio di richieste HTTP e struttura i tuoi modelli secondo API reali.
- Sviluppatore intermedio: Dai un’occhiata a meccanismi di nuova tentativo avanzati e strategie di ottimizzazione delle prestazioni. Considera di creare un metodo di nuova tentativo personalizzato per maggiore flessibilità.
- Sviluppatore senior: Dedica del tempo a perfezionare la gestione degli errori e i modelli di osservatore che possono fornire informazioni migliori sul comportamento della tua applicazione sotto carico. Le metriche e la registrazione sono tuoi amici!
Dati al 20 marzo 2026. Fonti: pydantic/pydantic-ai, Pydantic AI Retries, Aggiungi un comportamento personalizzato di nuova tentativo delle richieste.
Articoli correlati
- Midjourney News Ottobre 2025: La prossima avanzata dell’arte IA rivelata!
- La mia azienda locale LLMs con RAG
- Il mio motore di query LlamaIndex gestisce dati complessi
🕒 Published: