FastAPI で為替レート API を構築する方法（非同期 httpx・キャッシュ・Pydantic）

Python でフィンテックのバックエンド、複数通貨のチェックアウト、あるいは社内の価格計算サービスを構築していると、遅かれ早かれ自前の為替エンドポイントが必要になります。FastAPI で為替レート API を構築すると、ライブのレートプロバイダーをラップし、無料枠内に収めるためのキャッシュを追加し、他のサービスから呼び出せる整った /convert エンドポイントを公開する、高速で非同期かつ完全に型付けされたマイクロサービスが手に入ります。FastAPI はここに自然に適合します。本物の非同期 I/O のために ASGI 上に構築され、Pydantic でリクエストとレスポンスを検証し、対話的な OpenAPI ドキュメントを無料で生成します。このチュートリアルは、プロバイダーへの最初のリクエストから、共有 HTTP クライアント、キャッシュされた読み取り、Decimal による安全な金額処理、型付けされたエラー処理を備えた本番対応のコンバーターまで、道のり全体をたどります。

最後には、Finexly API ドキュメントと 170 以上の通貨カバレッジに支えられた、小さく自己完結した FastAPI サービスが手に入ります。すでに Python 通貨 API チュートリアルを読んでいれば、これは各部品が実際の ASGI バックエンドでどう組み合わさるかを示す、サービスレベルの姉妹編です。Django チュートリアルが ORM とテンプレート層に依拠するのに対し、FastAPI はすべてを軽量かつ非同期優先に保ちます。

なぜ FastAPI は通貨マイクロサービスに最適なのか

通貨換算は些細に見えます——金額にレートを掛けるだけ——が、それをサービスとしてきちんと行うことは、FastAPI がまさに設計対象とした関心事に触れます。

• 本物の非同期 I/O。 レート取得はネットワーク律速です。async def エンドポイントと非同期 HTTP クライアントを使えば、単一のワーカーが、呼び出しごとにスレッドをブロックする代わりに、リクエストが進行中のあいだ多くの同時換算を処理できます。
• 型付けされた契約。 Pydantic モデルは受信クエリパラメータを検証し、送信 JSON を形作るので、不正な amount や未知の通貨コードは、あなたのロジックに届く前に明確な 422 で拒否されます。
• 無料の対話的ドキュメント。 書いた各エンドポイントは /docs の Swagger UI に表示され、サービスをチームの他のメンバーにとって自己文書化されたものにします。
• lifespan 管理。 FastAPI は、起動時に単一の HTTP 接続プールを開き、シャットダウン時にそれを排出するための整った場所を提供します——性能にとって重要で、以下で見るとおりです。

私たちが構築するアーキテクチャは、薄くキャッシュを意識したプロキシです。サービスはレート API を呼び出し、短い期間レスポンスをキャッシュし、換算結果を自前のフロントエンドや他のマイクロサービスに提供します。このパターンは API キーをサーバー側に保ち、プロバイダーのレート制限からあなたを隔離し、ビジネスルール（丸め、上乗せ、許可リスト）を一箇所に追加できるようにします。

前提条件とプロジェクトのセットアップ

Python 3.11 以降が必要です。仮想環境を作成し、依存関係をインストールします。

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

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

fastapi[standard] は Uvicorn（ASGI サーバー）と CLI を引き込みます。httpx は非同期 HTTP クライアントで、pydantic-settings は環境変数からの設定を扱います。

Finexly ダッシュボードで無料 API キーを取得し、バージョン管理に決して入らないようエクスポートします。

export FINEXLY_API_KEY="your_api_key_here"

プロジェクト構成を作成します。

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

ステップ 1：pydantic-settings による設定

設定を一つの型付けされた場所にまとめます。pydantic-settings は環境変数と .env ファイルを読み取り、必須値が欠けていれば声高に失敗します。

# 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

設定オブジェクトはインポート時に作成されるため、キーが欠けていると、最初のリクエストで失敗するのではなく、サービスの起動を止めます——デプロイパイプラインでまさに望むことです。

ステップ 2：lifespan による共有の非同期 HTTP クライアント

これはサービス全体で最も重要な性能上の決定です。リクエストごとに新しい httpx.AsyncClient を作成してはいけません。 各クライアントは接続プールを持ちます。呼び出しごとに一つ作ると keep-alive 接続を捨て、毎回新しい TLS ハンドシェイクを強制します。代わりに、アプリ起動時に一つのクライアントを開き、プロセスの生存期間を通じて再利用します。

現代的なやり方は FastAPI の lifespan コンテキストマネージャです。（古い @app.on_event("startup") と @app.on_event("shutdown") デコレータは FastAPI 0.93+ で lifespan に取って代わられ非推奨になりました。）async with を使えば、起動が中断されても、シャットダウン時に接続プールが排出されソケットが閉じることが保証されます。

# 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

これで各エンドポイントは依存関係を通じて同じプール済みクライアントを借り、誰も接続の開閉を考える必要がなくなります。

ステップ 3：型付けされたリクエストとレスポンスのモデル

Pydantic モデルは検証と自己文書化されたレスポンスを提供します。金額に Decimal を使う点に注意してください。二進浮動小数点は十進小数を正確に表現できず、金額の丸め誤差は珍事ではなく本物のバグです。

# 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 はこれらのモデルを OpenAPI スキーマに描画し、値を自動的に強制変換・検証するので、意味を成さない金額のリクエストはハンドラが動く前に拒否されます。

ステップ 4：小さな TTL キャッシュでレートを取得する

レートはミリ秒ごとに変わるわけではなく、リクエストごとにプロバイダー API を叩くのは遅い上にクォータを素早く使い果たす方法です。生存時間（TTL）付きの小さなインメモリキャッシュがこれをならします。単一プロセスのサービスなら、この辞書ベースのキャッシュで十分です。複数ワーカーの場合は、呼び出し箇所を変えずに後で Redis に置き換えます。

# 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

プロバイダーの失敗があなたの呼び出し側にとって意味のある HTTP ステータスコードにどう変換されるかに注目してください。プロバイダーからの 429 は再試行のヒント付きの 503 になり、認証失敗は 502 になるので、資格情報の詳細を下流に漏らすことは決してありません。通貨 API のキャッシュとエラー処理に関する併読ガイドが、これらのパターンをより深く掘り下げます。

ステップ 5：/convert と /rates エンドポイント

いまやエンドポイント自体は小さく、難しい仕事はすべて fetch_rates の中にあります。クエリパラメータを FastAPI の Query で検証し、十進で安全な計算を行い、型付けされたモデルを返します。

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

ローカルで実行します。

fastapi dev app/main.py

次に、対話的な Swagger UI のために http://127.0.0.1:8000/docs を開くか、エンドポイントを直接呼び出します。

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

これで動作する通貨マイクロサービスができました。自前の換算計算をまったく走らせたくなければ、Finexly はホスト型の通貨コンバーターと、直接呼び出せる換算エンドポイントも提供しています。

ステップ 6：十進精度と小数のない通貨についての注意

二つの金銭の落とし穴が、ほぼすべての通貨サービスを捕まえます。

1. 金額に二進浮動小数点を決して使わないこと。 浮動小数点演算では 0.1 + 0.2 は 0.3 ではありません。レートと金額を Decimal(str(value)) で解析し、上のように最終結果を量子化します。
2. すべての通貨が小数点以下 2 桁を持つわけではない。 ISO 4217 は通貨ごとに補助単位を定義します。JPY と KRW は小数点以下ゼロ桁、BHD と KWD は三桁です。.quantize(Decimal("0.01")) をハードコードすると、円やディナールを黙って誤って丸めます。本番サービスは通貨ごとに正しい指数を調べ、それに応じて量子化すべきです。

これを正しく行うことこそ、デモと、実際の取引の前に置けるものとの違いです。

ステップ 7：サービスをテストする

HTTP クライアントは依存関係として注入されるため、テストは簡単です——依存関係をモックで上書きすれば、ネットワークに触れることはありません。

# 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 は lifespan イベントを代わりに実行するので、共有クライアントは本番とまったく同じようにセットアップされ、破棄されます。

本番へ：スケーリングとレート制限

出荷前に計画しておくべきいくつかのこと。

• 複数ワーカー。 複数ワーカーの Uvicorn/Gunicorn の下では、インメモリキャッシュはプロセス単位です。共有キャッシュのために Redis に移行すれば、あるワーカーが取得したレートがすべてに役立ちます。fetch_rates という継ぎ目が、これを単一関数の変更にします。
• プロバイダークォータ。 キャッシュは第一の防衛線です。5 分の TTL なら、いくつ換算を提供しても、一つの基準通貨あたりのプロバイダー呼び出しは 1 時間に最大 12 回です。料金ページでプランの制限を確認し、TTL を自分の階層に合わせて設定してください。
• 過去レート。 請求書、レポート、監査証跡には、特定日のレートを問い合わせる /historical エンドポイントが欲しくなります。同じクライアントとキャッシュのパターンが当てはまります。キャッシュを (base, date) でインデックスするだけです。
• 可観測性。 キャッシュのヒット/ミス率とプロバイダーのレイテンシを記録し、実データで TTL を調整します。

まだプロバイダーを選んでいる段階なら、通貨 API を比較するページがカバレッジ、制限、価格を並べて示します。

よくある質問

FastAPI で requests ではなく httpx を使うのはなぜ？

requests は同期的でイベントループをブロックし、非同期フレームワークの目的を台無しにします。httpx は接続プール、タイムアウト、再試行トランスポートを備えた完全非同期のクライアント（httpx.AsyncClient）を提供するので、他のリクエストをブロックせずに FastAPI の async def エンドポイントに直接組み込めます。

httpx クライアントはリクエストごとに作るべき？ それとも一度だけ？

一度だけ。lifespan コンテキストマネージャで単一の httpx.AsyncClient を作り、すべてのリクエストで再利用します。リクエストごとのクライアントは接続プールを捨て、呼び出しごとに新しい TLS ハンドシェイクを強制します。これは負荷下で計測できるほど遅く、ソケットを枯渇させることもあります。

通貨 API のレート制限に達しないようにするには？

プロバイダーのレスポンスを短い TTL（たとえば 5 分）でキャッシュします。レートはゆっくり動くので、基準通貨ごとのキャッシュは数千のユーザー換算を 1 時間あたり数回のプロバイダー呼び出しに減らします。複数ワーカーのデプロイでは、キャッシュがプロセス間で共有されるよう Redis を使います。

小数桁数の異なる通貨はどう扱う？

Python の Decimal 型を使い、float は決して使わず、ISO 4217 に従って各通貨の正しい補助単位の桁数に結果を量子化します。JPY と KRW はゼロ桁、多くは 2 桁、BHD/KWD は 3 桁を使います。2 桁をハードコードするのではなく、通貨ごとに指数を調べてください。

このパターンを Flask や Django のような同期フレームワークで使える？

キャッシュと Decimal のパターンはそのまま転用できますが、共有クライアントと非同期エンドポイントの部分は FastAPI 固有です。Django については、フレームワークのキャッシュバックエンドと ORM を使う専用の Django 通貨 API チュートリアルを参照してください。

構築を始めよう

これで FastAPI に、完全で非同期、キャッシュを意識した通貨マイクロサービスが手に入りました——Pydantic で型付けされ、Decimal で安全、プロバイダー障害に強い。一度に読み切れるほど小さく、実際の本番サービスへ育つのに十分なほど構造化されています。

リアルタイムの為替レートをプロジェクトに統合する準備はできましたか？ 無料の Finexly API キーを取得——クレジットカード不要。170 以上の通貨をカバーする気前のよい無料枠から始め、トラフィックの増加に合わせてスケールしてください。まず探索したいですか？ 無料通貨 APIの概要を見るか、API ドキュメントに飛び込んでください。