Как создать API обменных курсов на FastAPI (асинхронный httpx, кэширование, Pydantic)

Если вы создаёте финтех-бэкенд, мультивалютную оплату или внутренний сервис ценообразования на Python, рано или поздно вам понадобится собственный эндпоинт обмена валют. Создание API обменных курсов на FastAPI даёт вам быстрый, асинхронный и полностью типизированный микросервис, который оборачивает провайдера курсов в реальном времени, добавляет кэширование, чтобы оставаться в рамках бесплатного тарифа, и предоставляет аккуратный эндпоинт /convert, который могут вызывать другие ваши сервисы. FastAPI здесь подходит естественно: он построен на ASGI для настоящего асинхронного ввода-вывода, валидирует запросы и ответы через Pydantic и бесплатно генерирует интерактивную документацию OpenAPI. Этот учебник проходит весь путь — от вашего первого запроса к провайдеру до готового к продакшену конвертера с общим HTTP-клиентом, кэшированными чтениями, безопасными для денег вычислениями через Decimal и типизированной обработкой ошибок.

В итоге у вас будет небольшой самодостаточный сервис FastAPI, опирающийся на документацию API Finexly и её покрытие более 170 валют. Если вы уже читали наш учебник по API валют на Python, это дополнение на уровне сервиса, показывающее, как части складываются в настоящем ASGI-бэкенде. Там, где учебник по Django опирается на ORM и слой шаблонов, FastAPI держит всё лёгким и ориентированным на async.

Почему FastAPI хорошо подходит для валютного микросервиса

Конвертация валют кажется тривиальной — умножить сумму на курс — но сделать это хорошо как сервис затрагивает задачи, для которых FastAPI и был спроектирован:

• Настоящий асинхронный ввод-вывод. Получение курсов ограничено сетью. С эндпоинтами async def и асинхронным HTTP-клиентом один воркер может обслуживать множество одновременных конвертаций, пока запросы в пути, вместо блокировки потока на каждый вызов.
• Типизированные контракты. Модели Pydantic валидируют входящие параметры запроса и формируют исходящий JSON, так что некорректный amount или неизвестный код валюты отклоняется понятной ошибкой 422 ещё до того, как достигнет вашей логики.
• Бесплатная интерактивная документация. Каждый написанный вами эндпоинт появляется в Swagger UI по адресу /docs, что делает сервис самодокументируемым для остальной команды.
• Управление 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 управляет конфигурацией из переменных окружения.

Получите бесплатный API-ключ в панели Finexly и экспортируйте его, чтобы он никогда не попадал в систему контроля версий:

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: Общий асинхронный HTTP-клиент через lifespan

Это самое важное решение о производительности во всём сервисе. Не создавайте новый httpx.AsyncClient на каждый запрос. Каждый клиент владеет пулом соединений; создание одного на вызов выбрасывает keep-alive-соединения и вынуждает новый TLS-хендшейк каждый раз. Вместо этого откройте один клиент при старте приложения и переиспользуйте его на протяжении всей жизни процесса.

Современный способ — это контекстный менеджер lifespan от FastAPI. (Старые декораторы @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. Мы валидируем параметры запроса через Query FastAPI, делаем безопасную для десятичных чисел математику и возвращаем типизированные модели.

@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. Никогда не используйте двоичные float для сумм. 0.1 + 0.2 не равно 0.3 в арифметике с плавающей точкой. Разбирайте курсы и суммы через Decimal(str(value)) и квантуйте итоговый результат, как выше.
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 делает это изменением одной функции.
• Квота провайдера. Кэширование — ваша первая линия обороны. С TTL в 5 минут одна базовая валюта стоит максимум 12 вызовов провайдера в час, независимо от того, сколько конвертаций вы обслуживаете. Проверьте лимиты вашего тарифа на странице цен и настройте TTL под ваш уровень.
• Исторические курсы. Для счетов, отчётов или аудиторских следов вам понадобится эндпоинт /historical, который запрашивает курсы на конкретную дату. Применяются те же шаблоны клиента и кэша; просто индексируйте кэш по (base, date).
• Наблюдаемость. Логируйте соотношения попаданий/промахов кэша и задержку провайдера, чтобы настраивать TTL по реальным данным.

Если вы всё ещё выбираете провайдера, наша страница для сравнения API валют раскладывает покрытие, лимиты и цены рядом.

Часто задаваемые вопросы

Почему использовать httpx вместо requests в FastAPI?

requests синхронен и блокирует цикл событий, что сводит на нет смысл асинхронного фреймворка. httpx предлагает полностью асинхронный клиент (httpx.AsyncClient) с пулом соединений, таймаутами и транспортами с повторами, поэтому он напрямую вставляется в эндпоинты async def FastAPI, не блокируя другие запросы.

Создавать клиент httpx на каждый запрос или один раз?

Один раз. Создайте один httpx.AsyncClient в контекстном менеджере lifespan и переиспользуйте его во всех запросах. Клиент на запрос выбрасывает пул соединений и вынуждает новый TLS-хендшейк на каждый вызов, что заметно медленнее под нагрузкой и может исчерпать сокеты.

Как избежать достижения лимита запросов API валют?

Кэшируйте ответы провайдера с коротким TTL (например, 5 минут). Поскольку курсы двигаются медленно, кэширование по базовой валюте сокращает тысячи пользовательских конвертаций до горстки вызовов провайдера в час. Для многоворкерных развёртываний используйте Redis, чтобы кэш был общим между процессами.

Как обрабатывать валюты с разным числом десятичных знаков?

Используйте тип Decimal в Python, никогда не float, и квантуйте результат до правильного числа минорных единиц каждой валюты по ISO 4217. У JPY и KRW ноль знаков, у большинства — два, у BHD/KWD — три. Ищите показатель степени для каждой валюты вместо жёсткого задания двух знаков.

Можно ли использовать этот шаблон для синхронного фреймворка вроде Flask или Django?

Шаблоны кэширования и Decimal переносятся напрямую, но части с общим клиентом и асинхронным эндпоинтом специфичны для FastAPI. Для Django смотрите наш отдельный учебник по API валют на Django, который использует бэкенд кэша и ORM фреймворка.

Начните создавать

Теперь у вас есть полный, асинхронный, осведомлённый о кэше валютный микросервис на FastAPI — типизированный через Pydantic, безопасный с Decimal и устойчивый к сбоям провайдера. Он достаточно мал, чтобы прочитать за один присест, и достаточно структурирован, чтобы вырасти в настоящий продакшен-сервис.

Готовы интегрировать курсы валют в реальном времени в свой проект? Получите бесплатный API-ключ Finexly — без кредитной карты. Начните с щедрого бесплатного тарифа, покрывающего более 170 валют, и масштабируйтесь по мере роста трафика. Предпочитаете сначала осмотреться? Просмотрите обзор бесплатного API валют или погрузитесь в документацию API.