Come creare un'API dei tassi di cambio con FastAPI (httpx asincrono, caching, Pydantic)

Se stai costruendo un backend fintech, un checkout multivaluta o un servizio di prezzi interno in Python, prima o poi avrai bisogno di un tuo endpoint di cambio. Creare un'API dei tassi di cambio con FastAPI ti dà un microservizio veloce, asincrono e completamente tipizzato che avvolge un provider di tassi in tempo reale, aggiunge una cache per restare nel piano gratuito ed espone un endpoint /convert pulito che i tuoi altri servizi possono chiamare. FastAPI è una scelta naturale qui: è costruito su ASGI per I/O asincrono reale, valida richieste e risposte con Pydantic e genera gratuitamente documentazione OpenAPI interattiva. Questo tutorial percorre l'intero cammino — dalla tua prima richiesta al provider fino a un convertitore pronto per la produzione con un client HTTP condiviso, letture in cache, denaro sicuro con i decimali e gestione degli errori tipizzata.

Alla fine avrai un piccolo servizio FastAPI autonomo, sostenuto dalla documentazione dell'API Finexly e dalla sua copertura di oltre 170 valute. Se hai già letto il nostro tutorial dell'API delle valute in Python, questo è il complemento a livello di servizio che mostra come i pezzi si incastrano in un vero backend ASGI. Mentre il tutorial di Django si appoggia sull'ORM e sul livello dei template, FastAPI mantiene tutto snello e orientato all'async.

Perché FastAPI è un'ottima scelta per un microservizio di valute

La conversione di valute sembra banale — moltiplicare un importo per un tasso — ma farla bene come servizio tocca aspetti per cui FastAPI è stato progettato:

• I/O asincrono reale. Recuperare i tassi dipende dalla rete. Con endpoint async def e un client HTTP asincrono, un singolo worker può servire molte conversioni concorrenti mentre le richieste sono in corso, invece di bloccare un thread per chiamata.
• Contratti tipizzati. I modelli Pydantic validano i parametri di query in ingresso e modellano il JSON in uscita, così un amount malformato o un codice valuta sconosciuto viene rifiutato con un chiaro 422 prima ancora di raggiungere la tua logica.
• Documentazione interattiva gratuita. Ogni endpoint che scrivi appare nella Swagger UI su /docs, il che rende il servizio auto-documentato per il resto del tuo team.
• Gestione del lifespan. FastAPI ti offre un punto pulito per aprire un unico pool di connessioni HTTP all'avvio e svuotarlo allo spegnimento — cruciale per le prestazioni, come vedrai più avanti.

L'architettura che costruiamo è un proxy sottile e consapevole della cache: il tuo servizio chiama l'API dei tassi, mette la risposta in cache per una breve finestra e serve conversioni al tuo frontend o ad altri microservizi. Questo schema mantiene le tue chiavi API lato server, ti isola dai limiti di frequenza del provider e ti permette di aggiungere regole di business (arrotondamento, ricarichi, liste consentite) in un unico punto.

Prerequisiti e configurazione del progetto

Ti serve Python 3.11 o più recente. Crea un ambiente virtuale e installa le dipendenze:

python -m venv .venv
source .venv/bin/activate  # on Windows: .venv\Scripts\activate

pip install "fastapi[standard]" httpx pydantic-settings

fastapi[standard] include Uvicorn (il server ASGI) e la CLI. httpx è il nostro client HTTP asincrono, e pydantic-settings gestisce la configurazione dalle variabili d'ambiente.

Prendi una chiave API gratuita dalla tua dashboard Finexly ed esportala in modo che non finisca mai nel controllo di versione:

export FINEXLY_API_KEY="your_api_key_here"

Crea la struttura del progetto:

fx-service/
├── app/
│   ├── __init__.py
│   ├── config.py
│   ├── client.py
│   ├── models.py
│   └── main.py
└── .env

Passo 1: Configurazione con pydantic-settings

Tieni la configurazione in un unico posto tipizzato. pydantic-settings legge le variabili d'ambiente e un file .env, e fallisce rumorosamente se manca un valore richiesto.

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="FINEXLY_")

    api_key: str
    base_url: str = "https://api.finexly.com/v1"
    cache_ttl_seconds: int = 300
    request_timeout_seconds: float = 10.0


settings = Settings()  # raises at startup if FINEXLY_API_KEY is unset

Poiché l'oggetto di configurazione viene creato al momento dell'import, una chiave mancante impedisce l'avvio del servizio invece di fallire alla prima richiesta — esattamente ciò che vuoi in una pipeline di deployment.

Passo 2: Un client HTTP asincrono condiviso tramite lifespan

Questa è la decisione di prestazioni più importante dell'intero servizio. Non creare un nuovo httpx.AsyncClient per ogni richiesta. Ogni client possiede un pool di connessioni; crearne uno per chiamata butta via le connessioni keep-alive e forza un nuovo handshake TLS ogni volta. Apri invece un client all'avvio dell'app e riutilizzalo per tutta la vita del processo.

Il modo moderno per farlo è il gestore di contesto lifespan di FastAPI. (I vecchi decoratori @app.on_event("startup") e @app.on_event("shutdown") sono stati deprecati in FastAPI 0.93+ a favore del lifespan.) Usare async with garantisce che il pool di connessioni venga svuotato e i socket chiusi allo spegnimento, anche se l'avvio viene interrotto.

# app/client.py
import httpx
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request

from .config import settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Runs once on startup: open a single pooled client
    async with httpx.AsyncClient(
        base_url=settings.base_url,
        headers={"Authorization": f"Bearer {settings.api_key}"},
        timeout=settings.request_timeout_seconds,
    ) as client:
        app.state.http = client
        yield
    # On shutdown, the async-with block closes the pool automatically


def get_client(request: Request) -> httpx.AsyncClient:
    # FastAPI dependency: hand endpoints the shared client
    return request.app.state.http

Ora ogni endpoint prende in prestito lo stesso client del pool tramite una dependency, e nessuno deve pensare ad aprire o chiudere connessioni.

Passo 3: Modelli tipizzati di richiesta e risposta

I modelli Pydantic ti danno validazione e risposte auto-documentate. Nota l'uso di Decimal per gli importi monetari: i float binari non possono rappresentare esattamente le frazioni decimali, e gli errori di arrotondamento sul denaro sono un vero bug, non una curiosità.

# app/models.py
from decimal import Decimal
from pydantic import BaseModel, Field


class ConversionResult(BaseModel):
    base: str = Field(..., examples=["USD"])
    quote: str = Field(..., examples=["EUR"])
    amount: Decimal = Field(..., examples=["100.00"])
    rate: Decimal = Field(..., examples=["0.92"])
    converted: Decimal = Field(..., examples=["92.00"])
    as_of: str = Field(..., description="Timestamp of the upstream rate")


class RatesResponse(BaseModel):
    base: str
    rates: dict[str, Decimal]
    as_of: str

FastAPI renderizzerà questi modelli nello schema OpenAPI e convertirà/validerà i valori automaticamente, così una richiesta con un importo senza senso viene rifiutata prima che il tuo handler venga eseguito.

Passo 4: Recuperare i tassi con una piccola cache TTL

I tassi non cambiano da millisecondo a millisecondo, e martellare l'API del provider a ogni richiesta è sia lento sia un modo veloce di bruciare la tua quota. Una piccola cache in memoria con tempo di vita (TTL) attenua questo. Per un servizio a processo singolo questa cache basata su dizionario è sufficiente; per più worker, sostituiscila con Redis in seguito senza cambiare i punti di chiamata.

# app/main.py
import time
from decimal import Decimal, ROUND_HALF_UP

import httpx
from fastapi import Depends, FastAPI, HTTPException, Query

from .client import lifespan, get_client
from .config import settings
from .models import ConversionResult

app = FastAPI(title="FX Service", lifespan=lifespan)

# Simple in-memory cache: {base_currency: (expiry_timestamp, rates_dict)}
_cache: dict[str, tuple[float, dict]] = {}


async def fetch_rates(base: str, client: httpx.AsyncClient) -> dict:
    base = base.upper()
    now = time.monotonic()
    cached = _cache.get(base)
    if cached and cached[0] > now:
        return cached[1]  # cache hit — no network call

    try:
        resp = await client.get("/latest", params={"base": base})
        resp.raise_for_status()
    except httpx.HTTPStatusError as exc:
        code = exc.response.status_code
        if code == 429:
            raise HTTPException(503, "Upstream rate limit reached; try again shortly")
        if code in (401, 403):
            raise HTTPException(502, "Upstream authentication failed")
        raise HTTPException(502, "Upstream rates provider error")
    except httpx.RequestError:
        raise HTTPException(504, "Could not reach rates provider")

    rates = resp.json()["rates"]
    _cache[base] = (now + settings.cache_ttl_seconds, rates)
    return rates

Nota come i fallimenti del provider vengano tradotti in codici di stato HTTP significativi per i tuoi chiamanti. Un 429 dal provider diventa un 503 con un suggerimento di ritentare; un fallimento di autenticazione diventa un 502 così da non trapelare mai dettagli delle credenziali a valle. La nostra guida complementare su caching e gestione degli errori nelle API delle valute approfondisce questi schemi.

Passo 5: Gli endpoint /convert e /rates

Ora gli endpoint stessi sono minuscoli, perché tutto il lavoro difficile vive in fetch_rates. Validiamo i parametri di query con il Query di FastAPI, facciamo matematica sicura con i decimali e restituiamo modelli tipizzati.

@app.get("/convert", response_model=ConversionResult)
async def convert(
    base: str = Query(..., min_length=3, max_length=3),
    quote: str = Query(..., min_length=3, max_length=3),
    amount: Decimal = Query(..., gt=0),
    client: httpx.AsyncClient = Depends(get_client),
):
    base, quote = base.upper(), quote.upper()
    rates = await fetch_rates(base, client)

    if quote not in rates:
        raise HTTPException(400, f"Unsupported currency: {quote}")

    rate = Decimal(str(rates[quote]))
    converted = (amount * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

    return ConversionResult(
        base=base, quote=quote, amount=amount,
        rate=rate, converted=converted, as_of="latest",
    )


@app.get("/health")
async def health():
    return {"status": "ok"}

Eseguilo in locale:

fastapi dev app/main.py

Poi apri http://127.0.0.1:8000/docs per la Swagger UI interattiva, oppure chiama l'endpoint direttamente:

curl "http://127.0.0.1:8000/convert?base=USD&quote=EUR&amount=100"

{
  "base": "USD",
  "quote": "EUR",
  "amount": "100",
  "rate": "0.92",
  "converted": "92.00",
  "as_of": "latest"
}

Ora hai un microservizio di valute funzionante. Se preferisci non eseguire la tua matematica di conversione, Finexly offre anche un convertitore di valute ospitato e un endpoint di conversione diretto che puoi chiamare.

Passo 6: Una nota su precisione decimale e valute senza decimali

Due trappole del denaro colpiscono quasi ogni servizio di valute:

1. Non usare mai float binari per gli importi. 0.1 + 0.2 non è 0.3 nell'aritmetica in virgola mobile. Analizza tassi e importi tramite Decimal(str(value)) e quantizza il risultato finale, come fatto sopra.
2. Non tutte le valute hanno due decimali. L'ISO 4217 definisce unità minori per valuta: JPY e KRW hanno zero decimali, mentre BHD e KWD ne hanno tre. Codificare a mano .quantize(Decimal("0.01")) arrotonderà male lo yen e il dinaro in silenzio. Un servizio di produzione dovrebbe cercare l'esponente corretto per valuta e quantizzare di conseguenza.

Farlo bene è la differenza tra una demo e qualcosa che puoi mettere davanti a transazioni reali.

Passo 7: Testare il servizio

Poiché il client HTTP è iniettato come dependency, i test sono facili — sovrascrivi la dependency con un mock e non tocchi mai la rete:

# test_convert.py
from fastapi.testclient import TestClient
from app.main import app, fetch_rates

async def fake_rates(base, client):
    return {"EUR": "0.92", "GBP": "0.79"}

def test_convert(monkeypatch):
    monkeypatch.setattr("app.main.fetch_rates", fake_rates)
    with TestClient(app) as c:
        r = c.get("/convert?base=USD&quote=EUR&amount=100")
        assert r.status_code == 200
        assert r.json()["converted"] == "92.00"

TestClient esegue gli eventi di lifespan per te, quindi il client condiviso viene configurato e smontato esattamente come in produzione.

Verso la produzione: scalabilità e limiti di frequenza

Alcune cose da pianificare prima di rilasciare:

• Più worker. Con Uvicorn/Gunicorn e diversi worker, la tua cache in memoria è per processo. Passa a Redis per una cache condivisa, così che un tasso recuperato da un worker serva tutti. La separazione in fetch_rates rende questo un cambiamento di una singola funzione.
• Quota del provider. La cache è la tua prima linea di difesa. Con un TTL di 5 minuti, una valuta base costa al massimo 12 chiamate al provider all'ora, indipendentemente da quante conversioni servi. Controlla i limiti del tuo piano nella pagina dei prezzi e imposta il TTL adatto al tuo livello.
• Tassi storici. Per fatture, report o tracce di audit vorrai un endpoint /historical che interroghi i tassi di una data specifica. Si applicano gli stessi schemi di client e cache; indicizza semplicemente la cache per (base, date).
• Osservabilità. Registra i rapporti di hit/miss della cache e la latenza del provider per regolare il TTL con dati reali.

Se stai ancora scegliendo un provider, la nostra pagina per confrontare le API delle valute presenta copertura, limiti e prezzi affiancati.

Domande frequenti

Perché usare httpx invece di requests in FastAPI?

requests è sincrono e blocca l'event loop, il che vanifica lo scopo di un framework asincrono. httpx offre un client completamente asincrono (httpx.AsyncClient) con pool di connessioni, timeout e trasporti con ritentativi, così si integra direttamente negli endpoint async def di FastAPI senza bloccare altre richieste.

Devo creare il client httpx per richiesta o una sola volta?

Una sola volta. Crea un unico httpx.AsyncClient nel gestore di contesto lifespan e riutilizzalo in tutte le richieste. Un client per richiesta butta via il pool di connessioni e forza un nuovo handshake TLS a ogni chiamata, il che è misurabilmente più lento sotto carico e può esaurire i socket.

Come evito di raggiungere il limite di frequenza dell'API delle valute?

Metti in cache le risposte del provider con un TTL breve (per esempio 5 minuti). Poiché i tassi si muovono lentamente, la cache per valuta base riduce migliaia di conversioni utente a una manciata di chiamate al provider all'ora. Per deployment multi-worker, usa Redis così che la cache sia condivisa tra i processi.

Come gestisco valute con un numero diverso di decimali?

Usa il tipo Decimal di Python, mai float, e quantizza il risultato al numero corretto di unità minori di ogni valuta secondo l'ISO 4217. JPY e KRW usano zero decimali, la maggior parte ne usa due, e BHD/KWD ne usano tre. Cerca l'esponente per valuta invece di codificare a mano due decimali.

Posso usare questo schema per un framework sincrono come Flask o Django?

Gli schemi di cache e Decimal si trasferiscono direttamente, ma le parti con client condiviso ed endpoint asincrono sono specifiche di FastAPI. Per Django, vedi il nostro tutorial dedicato dell'API delle valute in Django, che usa il backend di cache e l'ORM del framework.

Inizia a costruire

Ora hai un microservizio di valute completo, asincrono e consapevole della cache in FastAPI — tipizzato con Pydantic, sicuro con Decimal e resiliente ai fallimenti del provider. È abbastanza piccolo da leggersi in una sola seduta e abbastanza strutturato da crescere fino a un vero servizio di produzione.

Pronto a integrare tassi di cambio in tempo reale nel tuo progetto? Ottieni la tua chiave API Finexly gratuita — senza carta di credito. Inizia con un generoso piano gratuito che copre oltre 170 valute e scala man mano che il tuo traffico cresce. Preferisci esplorare prima? Dai un'occhiata alla panoramica dell'API delle valute gratuita o immergiti nella documentazione dell'API.