Wechselkurs-API mit FastAPI erstellen (asynchrones httpx, Caching, Pydantic)

Wenn Sie ein Fintech-Backend, einen Multi-Währungs-Checkout oder einen internen Preisdienst in Python bauen, brauchen Sie früher oder später Ihren eigenen Wechselkurs-Endpunkt. Eine Wechselkurs-API mit FastAPI zu erstellen, gibt Ihnen einen schnellen, asynchronen und vollständig typisierten Microservice, der einen Live-Kursanbieter umhüllt, Caching hinzufügt, damit Sie im kostenlosen Kontingent bleiben, und einen sauberen /convert-Endpunkt bereitstellt, den Ihre anderen Dienste aufrufen können. FastAPI passt hier natürlich: Es basiert auf ASGI für echtes asynchrones I/O, validiert Anfragen und Antworten mit Pydantic und erzeugt kostenlos interaktive OpenAPI-Dokumentation. Dieses Tutorial geht den ganzen Weg — von Ihrer ersten Anfrage an den Anbieter bis zu einem produktionsreifen Konverter mit einem gemeinsam genutzten HTTP-Client, gecachten Lesevorgängen, dezimalsicherem Geld und typisierter Fehlerbehandlung.

Am Ende haben Sie einen kleinen, eigenständigen FastAPI-Dienst, gestützt auf die Finexly-API-Dokumentation und ihre Abdeckung von über 170 Währungen. Wenn Sie unser Python-Tutorial zur Währungs-API bereits gelesen haben, ist dies das Gegenstück auf Dienstebene, das zeigt, wie die Teile in einem echten ASGI-Backend zusammenpassen. Während das Django-Tutorial auf das ORM und die Template-Schicht setzt, hält FastAPI alles schlank und async-orientiert.

Warum FastAPI gut für einen Währungs-Microservice geeignet ist

Währungsumrechnung wirkt trivial — einen Betrag mit einem Kurs multiplizieren — aber sie gut als Dienst zu machen, berührt Aspekte, für die FastAPI entworfen wurde:

• Echtes asynchrones I/O. Kurse abzurufen ist netzwerkgebunden. Mit async def-Endpunkten und einem asynchronen HTTP-Client kann ein einzelner Worker viele gleichzeitige Umrechnungen bedienen, während Anfragen laufen, statt pro Aufruf einen Thread zu blockieren.
• Typisierte Verträge. Pydantic-Modelle validieren eingehende Query-Parameter und formen das ausgehende JSON, sodass ein fehlerhaftes amount oder ein unbekannter Währungscode mit einem klaren 422 abgelehnt wird, bevor es Ihre Logik erreicht.
• Kostenlose interaktive Dokumentation. Jeder Endpunkt, den Sie schreiben, erscheint in der Swagger UI unter /docs, was den Dienst für den Rest Ihres Teams selbstdokumentierend macht.
• Lifespan-Verwaltung. FastAPI bietet Ihnen einen sauberen Ort, um beim Start einen einzigen HTTP-Verbindungspool zu öffnen und beim Herunterfahren zu leeren — entscheidend für die Performance, wie Sie unten sehen werden.

Die Architektur, die wir bauen, ist ein dünner, cache-bewusster Proxy: Ihr Dienst ruft die Kurs-API auf, cacht die Antwort für ein kurzes Fenster und liefert Umrechnungen an Ihr eigenes Frontend oder andere Microservices. Dieses Muster hält Ihre API-Schlüssel serverseitig, schirmt Sie von den Ratenbegrenzungen des Anbieters ab und erlaubt Ihnen, Geschäftsregeln (Rundung, Aufschläge, Allowlists) an einer Stelle hinzuzufügen.

Voraussetzungen und Projekt-Setup

Sie benötigen Python 3.11 oder neuer. Erstellen Sie eine virtuelle Umgebung und installieren Sie die Abhängigkeiten:

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

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

fastapi[standard] bringt Uvicorn (den ASGI-Server) und die CLI mit. httpx ist unser asynchroner HTTP-Client, und pydantic-settings verwaltet die Konfiguration aus Umgebungsvariablen.

Holen Sie sich einen kostenlosen API-Schlüssel in Ihrem Finexly-Dashboard und exportieren Sie ihn, damit er nie in der Versionskontrolle landet:

export FINEXLY_API_KEY="your_api_key_here"

Legen Sie die Projektstruktur an:

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

Schritt 1: Konfiguration mit pydantic-settings

Halten Sie die Konfiguration an einem einzigen typisierten Ort. pydantic-settings liest Umgebungsvariablen und eine .env-Datei und schlägt lautstark fehl, wenn ein erforderlicher Wert fehlt.

# 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

Da das Konfigurationsobjekt zur Importzeit erstellt wird, verhindert ein fehlender Schlüssel den Start des Dienstes, statt bei der ersten Anfrage zu scheitern — genau das, was Sie in einer Deployment-Pipeline wollen.

Schritt 2: Ein gemeinsam genutzter, asynchroner HTTP-Client über Lifespan

Dies ist die wichtigste Performance-Entscheidung im gesamten Dienst. Erstellen Sie keinen neuen httpx.AsyncClient pro Anfrage. Jeder Client besitzt einen Verbindungspool; einen pro Aufruf zu erstellen, verwirft die Keep-Alive-Verbindungen und erzwingt jedes Mal einen neuen TLS-Handshake. Öffnen Sie stattdessen einen Client beim App-Start und verwenden Sie ihn für die gesamte Lebensdauer des Prozesses wieder.

Der moderne Weg dafür ist der Lifespan-Kontextmanager von FastAPI. (Die älteren Dekoratoren @app.on_event("startup") und @app.on_event("shutdown") wurden in FastAPI 0.93+ zugunsten von Lifespan als veraltet markiert.) async with zu verwenden, garantiert, dass der Verbindungspool geleert und die Sockets beim Herunterfahren geschlossen werden, selbst wenn der Start unterbrochen wird.

# 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

Jetzt leiht sich jeder Endpunkt denselben gepoolten Client über eine Dependency, und niemand muss an das Öffnen oder Schließen von Verbindungen denken.

Schritt 3: Typisierte Request- und Response-Modelle

Pydantic-Modelle geben Ihnen Validierung und selbstdokumentierende Antworten. Beachten Sie die Verwendung von Decimal für Geldbeträge: Binäre Fließkommazahlen können Dezimalbrüche nicht exakt darstellen, und Rundungsfehler bei Geld sind ein echter Bug, keine Kuriosität.

# 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 diese Modelle im OpenAPI-Schema und wandelt/validiert die Werte automatisch, sodass eine Anfrage mit einem sinnlosen Betrag abgelehnt wird, bevor Ihr Handler läuft.

Schritt 4: Kurse mit einem kleinen TTL-Cache abrufen

Kurse ändern sich nicht von Millisekunde zu Millisekunde, und die Anbieter-API bei jeder Anfrage zu bombardieren ist sowohl langsam als auch ein schneller Weg, Ihr Kontingent aufzubrauchen. Ein kleiner In-Memory-Cache mit Lebensdauer (TTL) glättet das. Für einen Ein-Prozess-Dienst reicht dieser wörterbuchbasierte Cache; für mehrere Worker tauschen Sie ihn später gegen Redis, ohne die Aufrufstellen zu ändern.

# 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

Beachten Sie, wie Anbieterfehler in aussagekräftige HTTP-Statuscodes für Ihre Aufrufer übersetzt werden. Ein 429 vom Anbieter wird zu einem 503 mit Wiederholungshinweis; ein Authentifizierungsfehler wird zu einem 502, damit Sie niemals Zugangsdaten-Details nach unten durchsickern lassen. Unser begleitender Leitfaden zu Caching und Fehlerbehandlung bei Währungs-APIs vertieft diese Muster.

Schritt 5: Die Endpunkte /convert und /rates

Nun sind die Endpunkte selbst winzig, weil die ganze harte Arbeit in fetch_rates steckt. Wir validieren die Query-Parameter mit FastAPIs Query, rechnen dezimalsicher und geben typisierte Modelle zurück.

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

Lokal ausführen:

fastapi dev app/main.py

Öffnen Sie dann http://127.0.0.1:8000/docs für die interaktive Swagger UI oder rufen Sie den Endpunkt direkt auf:

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

Sie haben jetzt einen funktionierenden Währungs-Microservice. Wenn Sie lieber gar keine eigene Umrechnungsmathematik ausführen möchten, bietet Finexly auch einen gehosteten Währungsrechner und einen direkten Umrechnungs-Endpunkt, den Sie aufrufen können.

Schritt 6: Ein Hinweis zu Dezimalgenauigkeit und Währungen ohne Nachkommastellen

Zwei Geldfallen erwischen fast jeden Währungsdienst:

1. Verwenden Sie niemals binäre Fließkommazahlen für Beträge. 0.1 + 0.2 ist in der Fließkomma-Arithmetik nicht 0.3. Parsen Sie Kurse und Beträge über Decimal(str(value)) und quantisieren Sie das Endergebnis, wie oben gezeigt.
2. Nicht jede Währung hat zwei Nachkommastellen. ISO 4217 definiert Untereinheiten pro Währung: JPY und KRW haben null Nachkommastellen, während BHD und KWD drei haben. .quantize(Decimal("0.01")) fest zu verdrahten, rundet Yen und Dinar stillschweigend falsch. Ein Produktionsdienst sollte den korrekten Exponenten pro Währung nachschlagen und entsprechend quantisieren.

Das richtig hinzubekommen ist der Unterschied zwischen einer Demo und etwas, das Sie vor echte Transaktionen stellen können.

Schritt 7: Den Dienst testen

Da der HTTP-Client als Dependency injiziert wird, sind Tests einfach — überschreiben Sie die Dependency mit einem Mock, und Sie berühren nie das Netzwerk:

# 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 führt die Lifespan-Ereignisse für Sie aus, sodass der gemeinsame Client genau wie in der Produktion aufgesetzt und abgebaut wird.

Auf dem Weg in die Produktion: Skalierung und Ratenbegrenzungen

Ein paar Dinge, die Sie vor dem Ausrollen einplanen sollten:

• Mehrere Worker. Unter Uvicorn/Gunicorn mit mehreren Workern ist Ihr In-Memory-Cache pro Prozess. Wechseln Sie zu Redis für einen gemeinsamen Cache, damit ein von einem Worker abgerufener Kurs allen dient. Die Trennstelle fetch_rates macht das zu einer Änderung an einer einzigen Funktion.
• Anbieter-Kontingent. Caching ist Ihre erste Verteidigungslinie. Mit einem TTL von 5 Minuten kostet eine Basiswährung höchstens 12 Anbieteraufrufe pro Stunde, egal wie viele Umrechnungen Sie bedienen. Prüfen Sie die Grenzen Ihres Tarifs auf der Preisseite und stellen Sie den TTL passend zu Ihrer Stufe ein.
• Historische Kurse. Für Rechnungen, Berichte oder Prüfpfade werden Sie einen /historical-Endpunkt wollen, der Kurse für ein bestimmtes Datum abfragt. Es gelten dieselben Client- und Cache-Muster; indexieren Sie den Cache einfach nach (base, date).
• Beobachtbarkeit. Protokollieren Sie Cache-Trefferquoten und Anbieterlatenz, um den TTL mit echten Daten abzustimmen.

Wenn Sie noch einen Anbieter auswählen, stellt unsere Seite zum Vergleich von Währungs-APIs Abdeckung, Grenzen und Preise nebeneinander dar.

Häufig gestellte Fragen

Warum httpx statt requests in FastAPI verwenden?

requests ist synchron und blockiert die Event-Loop, was den Zweck eines asynchronen Frameworks zunichtemacht. httpx bietet einen vollständig asynchronen Client (httpx.AsyncClient) mit Verbindungspool, Timeouts und Retry-Transports, sodass er sich direkt in FastAPIs async def-Endpunkte einfügt, ohne andere Anfragen zu blockieren.

Sollte ich den httpx-Client pro Anfrage oder einmal erstellen?

Einmal. Erstellen Sie einen einzigen httpx.AsyncClient im Lifespan-Kontextmanager und verwenden Sie ihn über alle Anfragen hinweg wieder. Ein Client pro Anfrage verwirft den Verbindungspool und erzwingt bei jedem Aufruf einen neuen TLS-Handshake, was unter Last messbar langsamer ist und die Sockets erschöpfen kann.

Wie vermeide ich, die Ratenbegrenzung der Währungs-API zu erreichen?

Cachen Sie die Anbieterantworten mit einem kurzen TTL (zum Beispiel 5 Minuten). Da sich Kurse langsam bewegen, reduziert Caching pro Basiswährung tausende Nutzerumrechnungen auf eine Handvoll Anbieteraufrufe pro Stunde. Für Multi-Worker-Deployments verwenden Sie Redis, damit der Cache über Prozesse hinweg geteilt wird.

Wie gehe ich mit Währungen unterschiedlicher Nachkommastellen um?

Verwenden Sie Pythons Decimal-Typ, niemals float, und quantisieren Sie das Ergebnis auf die korrekte Anzahl von Untereinheiten jeder Währung gemäß ISO 4217. JPY und KRW verwenden null Nachkommastellen, die meisten zwei und BHD/KWD drei. Schlagen Sie den Exponenten pro Währung nach, statt zwei Nachkommastellen fest zu verdrahten.

Kann ich dieses Muster für ein synchrones Framework wie Flask oder Django verwenden?

Die Cache- und Decimal-Muster lassen sich direkt übertragen, aber die Teile mit gemeinsam genutztem Client und asynchronem Endpunkt sind FastAPI-spezifisch. Für Django siehe unser eigenes Django-Tutorial zur Währungs-API, das das Cache-Backend und das ORM des Frameworks nutzt.

Loslegen

Sie haben jetzt einen vollständigen, asynchronen, cache-bewussten Währungs-Microservice in FastAPI — typisiert mit Pydantic, sicher mit Decimal und widerstandsfähig gegen Anbieterausfälle. Er ist klein genug, um ihn in einer Sitzung zu lesen, und strukturiert genug, um zu einem echten Produktionsdienst zu wachsen.

Bereit, Echtzeit-Wechselkurse in Ihr Projekt zu integrieren? Holen Sie sich Ihren kostenlosen Finexly-API-Schlüssel — keine Kreditkarte erforderlich. Beginnen Sie mit einem großzügigen kostenlosen Kontingent für über 170 Währungen und skalieren Sie mit wachsendem Traffic. Möchten Sie lieber zuerst stöbern? Sehen Sie sich die Übersicht der kostenlosen Währungs-API an oder tauchen Sie in die API-Dokumentation ein.