Een wisselkoers-API bouwen met FastAPI (asynchrone httpx, caching, Pydantic)

Als je een fintech-backend, een multivaluta-checkout of een interne prijsservice in Python bouwt, heb je vroeg of laat je eigen wisselkoers-endpoint nodig. Het bouwen van een wisselkoers-API met FastAPI geeft je een snelle, asynchrone en volledig getypeerde microservice die een live-koersprovider omhult, caching toevoegt zodat je binnen een gratis niveau blijft, en een net /convert-endpoint blootstelt dat je andere services kunnen aanroepen. FastAPI past hier van nature: het is gebouwd op ASGI voor echte async-I/O, het valideert verzoeken en antwoorden met Pydantic en het genereert gratis interactieve OpenAPI-documentatie. Deze tutorial doorloopt het hele pad — van je eerste verzoek aan de provider tot een productieklare converter met een gedeelde HTTP-client, gecachte reads, decimaalveilig geld en getypeerde foutafhandeling.

Aan het eind heb je een kleine, op zichzelf staande FastAPI-service, ondersteund door de Finexly API-documentatie en de dekking van meer dan 170 valuta's. Als je onze Python-tutorial voor de valuta-API al hebt gelezen, is dit de tegenhanger op serviceniveau die laat zien hoe de stukken in een echte ASGI-backend in elkaar passen. Waar de Django-tutorial leunt op de ORM en de templatelaag, houdt FastAPI alles slank en async-gericht.

Waarom FastAPI goed past bij een valuta-microservice

Valutaconversie lijkt triviaal — een bedrag met een koers vermenigvuldigen — maar het goed doen als service raakt zorgen waarvoor FastAPI is ontworpen:

Echte async-I/O. Koersen ophalen is netwerkgebonden. Met async def-endpoints en een asynchrone HTTP-client kan één worker veel gelijktijdige conversies bedienen terwijl verzoeken onderweg zijn, in plaats van per aanroep een thread te blokkeren.
Getypeerde contracten. Pydantic-modellen valideren binnenkomende queryparameters en vormen de uitgaande JSON, zodat een misvormde amount of een onbekende valutacode met een duidelijke 422 wordt afgewezen voordat het je logica bereikt.
Gratis interactieve documentatie. Elk endpoint dat je schrijft verschijnt in Swagger UI op /docs, wat de service zelfdocumenterend maakt voor de rest van je team.
Lifespan-beheer. FastAPI geeft je een nette plek om bij het opstarten één HTTP-verbindingspool te openen en bij het afsluiten te legen — cruciaal voor prestaties, zoals je hieronder zult zien.

De architectuur die we bouwen is een dunne, cache-bewuste proxy: je service roept de koers-API aan, cachet het antwoord gedurende een kort venster en levert conversies aan je eigen frontend of andere microservices. Dat patroon houdt je API-sleutels aan de serverkant, schermt je af van de rate limits van de provider en laat je bedrijfsregels (afronding, opslagen, allowlists) op één plek toevoegen.

Vereisten en projectopzet

Je hebt Python 3.11 of nieuwer nodig. Maak een virtuele omgeving en installeer de afhankelijkheden:

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

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

fastapi[standard] brengt Uvicorn (de ASGI-server) en de CLI mee. httpx is onze asynchrone HTTP-client, en pydantic-settings regelt de configuratie vanuit omgevingsvariabelen.

Haal een gratis API-sleutel uit je Finexly-dashboard en exporteer die zodat hij nooit in versiebeheer belandt:

export FINEXLY_API_KEY="your_api_key_here"

Maak de projectstructuur:

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

Stap 1: Configuratie met pydantic-settings

Houd de configuratie op één getypeerde plek. pydantic-settings leest omgevingsvariabelen en een .env-bestand, en faalt luidruchtig als een vereiste waarde ontbreekt.

# 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

Omdat het configuratieobject bij het importeren wordt aangemaakt, voorkomt een ontbrekende sleutel dat de service opstart in plaats van te falen bij het eerste verzoek — precies wat je wilt in een deploymentpijplijn.

Stap 2: Een gedeelde, asynchrone HTTP-client via lifespan

Dit is de belangrijkste prestatiebeslissing in de hele service. Maak niet per verzoek een nieuwe httpx.AsyncClient aan. Elke client bezit een verbindingspool; er per aanroep één maken gooit de keep-alive-verbindingen weg en dwingt elke keer een nieuwe TLS-handshake af. Open in plaats daarvan één client wanneer de app start en hergebruik hem gedurende de hele levensduur van het proces.

De moderne manier is de lifespan-contextmanager van FastAPI. (De oudere decorators @app.on_event("startup") en @app.on_event("shutdown") zijn in FastAPI 0.93+ afgeschaft ten gunste van lifespan.) async with gebruiken garandeert dat de verbindingspool leegloopt en de sockets sluiten bij het afsluiten, zelfs als het opstarten wordt onderbroken.

# 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

Nu leent elk endpoint dezelfde gepoolde client via een dependency, en niemand hoeft na te denken over het openen of sluiten van verbindingen.

Stap 3: Getypeerde request- en responsmodellen

Pydantic-modellen geven je validatie en zelfdocumenterende antwoorden. Let op het gebruik van Decimal voor geldbedragen: binaire floats kunnen decimale breuken niet exact voorstellen, en afrondingsfouten bij geld zijn een echte bug, geen curiositeit.

# 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 rendert deze modellen in het OpenAPI-schema en coerceert/valideert de waarden automatisch, zodat een verzoek met een onzinnig bedrag wordt afgewezen voordat je handler draait.

Stap 4: Koersen ophalen met een kleine TTL-cache

Koersen veranderen niet van milliseconde tot milliseconde, en de provider-API bij elk verzoek bestoken is zowel traag als een snelle manier om je quotum op te branden. Een kleine in-memory-cache met levensduur (TTL) vlakt dit af. Voor een service met één proces is deze op een dictionary gebaseerde cache genoeg; voor meerdere workers vervang je hem later door Redis zonder de aanroeppunten te wijzigen.

# 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

Merk op hoe providerfouten worden vertaald naar betekenisvolle HTTP-statuscodes voor jouw aanroepers. Een 429 van de provider wordt een 503 met een hint om opnieuw te proberen; een authenticatiefout wordt een 502 zodat je nooit details van inloggegevens stroomafwaarts lekt. Onze aanvullende gids over caching en foutafhandeling bij valuta-API's gaat dieper in op deze patronen.

Stap 5: De /convert- en /rates-endpoints

Nu zijn de endpoints zelf piepklein, omdat al het zware werk in fetch_rates zit. We valideren de queryparameters met FastAPI's Query, doen decimaalveilige wiskunde en geven getypeerde modellen terug.

@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"}

Draai het lokaal:

fastapi dev app/main.py

Open dan http://127.0.0.1:8000/docs voor de interactieve Swagger UI, of roep het endpoint direct aan:

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"
}

Je hebt nu een werkende valuta-microservice. Als je liever je eigen conversiewiskunde niet uitvoert, biedt Finexly ook een gehoste valutaomrekenaar en een direct conversie-endpoint dat je kunt aanroepen.

Stap 6: Een opmerking over decimale precisie en valuta's zonder decimalen

Twee geldvalstrikken vangen bijna elke valutaservice:

Gebruik nooit binaire floats voor bedragen. 0.1 + 0.2 is geen 0.3 in floating-point-rekenkunde. Parse koersen en bedragen via Decimal(str(value)) en quantize het eindresultaat, zoals hierboven.
Niet elke valuta heeft twee decimalen. ISO 4217 definieert kleinere eenheden per valuta: JPY en KRW hebben nul decimalen, terwijl BHD en KWD er drie hebben. .quantize(Decimal("0.01")) hardcoderen zal yen en dinar stilletjes verkeerd afronden. Een productieservice zou de juiste exponent per valuta moeten opzoeken en dienovereenkomstig quantizen.

Dit goed doen is het verschil tussen een demo en iets dat je voor echte transacties kunt zetten.

Stap 7: De service testen

Omdat de HTTP-client als dependency wordt geïnjecteerd, zijn tests eenvoudig — overschrijf de dependency met een mock en je raakt nooit het netwerk aan:

# 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 voert de lifespan-events voor je uit, dus de gedeelde client wordt opgezet en afgebroken precies zoals in productie.

Naar productie: schalen en rate limits

Een paar zaken om te plannen voordat je live gaat:

Meerdere workers. Onder Uvicorn/Gunicorn met meerdere workers is je in-memory-cache per proces. Stap over op Redis voor een gedeelde cache, zodat een koers die door één worker is opgehaald voor allemaal dient. De naad in fetch_rates maakt dit een wijziging van één functie.
Providerquotum. Caching is je eerste verdedigingslinie. Met een TTL van 5 minuten kost één basisvaluta hoogstens 12 provideraanroepen per uur, ongeacht hoeveel conversies je bedient. Bekijk de limieten van je plan op de prijzenpagina en stel de TTL passend bij je niveau in.
Historische koersen. Voor facturen, rapportages of audittrails wil je een /historical-endpoint dat koersen voor een specifieke datum opvraagt. Dezelfde client- en cachepatronen gelden; indexeer de cache gewoon op (base, date).
Observability. Log de hit/miss-verhoudingen van de cache en de providerlatentie om de TTL met echte data af te stemmen.

Als je nog een provider kiest, zet onze pagina om valuta-API's te vergelijken dekking, limieten en prijzen naast elkaar.

Veelgestelde vragen

Waarom httpx gebruiken in plaats van requests in FastAPI?

requests is synchroon en blokkeert de event-loop, wat het doel van een asynchroon framework tenietdoet. httpx biedt een volledig asynchrone client (httpx.AsyncClient) met verbindingspool, timeouts en retry-transports, dus hij past direct in FastAPI's async def-endpoints zonder andere verzoeken te blokkeren.

Moet ik de httpx-client per verzoek of één keer aanmaken?

Één keer. Maak één httpx.AsyncClient in de lifespan-contextmanager en hergebruik hem over alle verzoeken. Een client per verzoek gooit de verbindingspool weg en dwingt bij elke aanroep een nieuwe TLS-handshake af, wat meetbaar trager is onder belasting en de sockets kan uitputten.

Hoe voorkom ik dat ik de rate limit van de valuta-API bereik?

Cache de providerantwoorden met een korte TTL (bijvoorbeeld 5 minuten). Omdat koersen langzaam bewegen, reduceert caching per basisvaluta duizenden gebruikersconversies tot een handjevol provideraanroepen per uur. Gebruik voor multi-worker-deployments Redis zodat de cache over processen wordt gedeeld.

Hoe ga ik om met valuta's met een verschillend aantal decimalen?

Gebruik Python's Decimal-type, nooit float, en quantize het resultaat naar het juiste aantal kleinere eenheden van elke valuta volgens ISO 4217. JPY en KRW gebruiken nul decimalen, de meeste gebruiken er twee, en BHD/KWD gebruiken er drie. Zoek de exponent per valuta op in plaats van twee decimalen te hardcoderen.

Kan ik dit patroon gebruiken voor een synchroon framework zoals Flask of Django?

De cache- en Decimal-patronen zijn direct overdraagbaar, maar de delen met gedeelde client en asynchroon endpoint zijn FastAPI-specifiek. Zie voor Django onze speciale Django-tutorial voor de valuta-API, die de cache-backend en de ORM van het framework gebruikt.

Begin met bouwen

Je hebt nu een complete, asynchrone, cache-bewuste valuta-microservice in FastAPI — getypeerd met Pydantic, veilig met Decimal en bestand tegen providerstoringen. Hij is klein genoeg om in één keer te lezen en gestructureerd genoeg om uit te groeien tot een echte productieservice.

Klaar om realtime wisselkoersen in je project te integreren? Haal je gratis Finexly API-sleutel — geen creditcard nodig. Begin met een royaal gratis niveau dat meer dan 170 valuta's dekt, en schaal op naarmate je verkeer groeit. Wil je liever eerst verkennen? Bekijk het overzicht van de gratis valuta-API of duik in de API-documentatie.