返回博客

如何用 FastAPI 构建汇率 API（异步 httpx、缓存、Pydantic）

V
Vlado Grigirov
August 11, 2026
FastAPI Python Currency API Exchange Rates Tutorial Fintech

如果你正在用 Python 构建金融科技后端、多币种结账或内部定价服务，迟早会需要自己的汇率端点。用 FastAPI 构建汇率 API 能为你提供一个快速、异步且完全类型化的微服务：它封装了一个实时汇率提供方，加入缓存以让你保持在免费额度内，并暴露一个整洁的 /convert 端点供你的其他服务调用。FastAPI 在这里天然合适：它基于 ASGI 实现真正的异步 I/O，用 Pydantic 校验请求和响应，并免费生成交互式 OpenAPI 文档。本教程走完整条路径——从你对提供方的第一次请求，到一个可用于生产的转换器，包含共享 HTTP 客户端、缓存读取、基于 Decimal 的安全金额处理，以及类型化的错误处理。

读到最后，你将拥有一个小巧、自包含的 FastAPI 服务，由 Finexly API 文档及其对 170 多种货币的覆盖支撑。如果你已经读过我们的 Python 货币 API 教程，这一篇是服务层面的姊妹篇，展示这些部件如何在真实的 ASGI 后端中拼合。Django 教程依赖 ORM 与模板层，而 FastAPI 则让一切保持精简、以异步为先。

为什么 FastAPI 非常适合货币微服务

货币转换看似微不足道——把金额乘以汇率——但把它作为服务做好，会触及 FastAPI 正是为之设计的诸多关切：

  • 真正的异步 I/O。 获取汇率是网络受限的。使用 async def 端点和异步 HTTP 客户端，单个 worker 可以在请求进行中同时处理许多并发转换，而不是每次调用阻塞一个线程。
  • 类型化契约。 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：二进制浮点数无法精确表示十进制小数，货币中的取整误差是真正的 bug，而非趣闻。

# 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）的小型内存缓存能平滑这一点。对于单进程服务，这个基于字典的缓存就够了；对于多个 worker，之后换成 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

然后打开 http://127.0.0.1:8000/docs 查看交互式 Swagger UI，或直接调用端点：

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. 并非每种货币都有两位小数。 ISO 4217 为每种货币定义了次级单位：JPY 和 KRW 有零位小数，而 BHD 和 KWD 有三位。硬编码 .quantize(Decimal("0.01")) 会悄悄地对日元和第纳尔进行错误取整。生产服务应按货币查找正确的指数并相应地量化。

把这一点做对，就是一个演示与你能放在真实交易前面的东西之间的区别。

第 7 步：测试该服务

由于 HTTP 客户端作为依赖注入，测试很容易——用一个 mock 覆盖该依赖，你就永远不会碰到网络：

# 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 事件，因此共享客户端的建立与拆除与生产环境完全一致。

走向生产：扩展与速率限制

上线前需要规划的几件事：

  • 多个 worker。 在带有多个 worker 的 Uvicorn/Gunicorn 下，你的内存缓存是按进程的。改用 Redis 以获得共享缓存，这样一个 worker 获取的汇率就能服务所有 worker。fetch_rates 这个接缝让这成为一次单函数的改动。
  • 提供方配额。 缓存是你的第一道防线。使用 5 分钟的 TTL，无论你服务多少次转换，一种基础货币每小时最多耗费 12 次提供方调用。在价格页面查看你套餐的限制，并把 TTL 设置得与你的档位相称。
  • 历史汇率。 对于发票、报表或审计线索，你会想要一个 /historical 端点来查询特定日期的汇率。同样的客户端与缓存模式依然适用；只需按 (base, date) 索引缓存。
  • 可观测性。 记录缓存命中/未命中比率和提供方延迟，以便用真实数据调优 TTL。

如果你还在选择提供方，我们的货币 API 对比页面把覆盖范围、限制和价格并排呈现。

常见问题

在 FastAPI 中为什么用 httpx 而不是 requests？

requests 是同步的，会阻塞事件循环，这抵消了异步框架的意义。httpx 提供了一个完全异步的客户端（httpx.AsyncClient），带有连接池、超时和重试传输，因此它能直接接入 FastAPI 的 async def 端点，而不会阻塞其他请求。

我应该按请求创建 httpx 客户端，还是只创建一次？

一次。在 lifespan 上下文管理器中创建单个 httpx.AsyncClient，并在所有请求间复用它。按请求创建的客户端会丢弃连接池，并在每次调用时强制新的 TLS 握手，这在负载下明显更慢，并可能耗尽套接字。

如何避免触及货币 API 的速率限制？

用较短的 TTL（例如 5 分钟）缓存提供方的响应。由于汇率变动缓慢，按基础货币缓存能把数千次用户转换减少为每小时少数几次提供方调用。对于多 worker 部署，使用 Redis，让缓存在进程间共享。

如何处理小数位数不同的货币？

使用 Python 的 Decimal 类型，绝不用 float，并按 ISO 4217 把结果量化到每种货币正确的次级单位位数。JPY 和 KRW 用零位小数，大多数用两位，BHD/KWD 用三位。按货币查找指数，而不是硬编码两位小数。

我能把这个模式用于像 Flask 或 Django 这样的同步框架吗？

缓存与 Decimal 模式可以直接迁移，但共享客户端与异步端点部分是 FastAPI 特有的。对于 Django，请参阅我们专门的 Django 货币 API 教程，它使用框架的缓存后端和 ORM。

开始构建

现在你在 FastAPI 中拥有了一个完整的、异步的、感知缓存的货币微服务——用 Pydantic 类型化，用 Decimal 保证安全，并能抵御提供方故障。它小到可以一口气读完，又结构化到足以成长为真正的生产服务。

准备好把实时汇率集成到你的项目中了吗？获取你的免费 Finexly API 密钥——无需信用卡。从覆盖 170 多种货币的慷慨免费档位开始，并随流量增长而扩展。想先探索一下？浏览免费货币 API概览，或深入 API 文档。

Explore More

API Documentation

Full endpoint reference & examples
Currency Converter

Try live rates instantly
Pricing Plans

Free tier & production plans
VG

Vlado Grigirov

Senior Currency Markets Analyst & Financial Strategist

Vlado Grigirov is a senior currency markets analyst and financial strategist with over 14 years of experience in foreign exchange markets, cross-border finance, and currency risk management. He has wo...

View full profile →

分享这篇文章

推特 领英 电子邮件