Como criar uma API de taxas de câmbio com FastAPI (httpx assíncrono, cache e Pydantic)

Se você está construindo um backend fintech, um checkout multimoeda ou um serviço interno de preços em Python, mais cedo ou mais tarde vai precisar do seu próprio endpoint de câmbio. Criar uma API de taxas de câmbio com FastAPI oferece um microsserviço rápido, assíncrono e totalmente tipado que envolve um provedor de taxas ao vivo, adiciona cache para você permanecer dentro de um plano gratuito e expõe um endpoint /convert limpo que seus outros serviços podem chamar. O FastAPI se encaixa naturalmente aqui: é construído sobre ASGI para I/O assíncrono real, valida requisições e respostas com Pydantic e gera documentação interativa OpenAPI de graça. Este tutorial percorre todo o caminho — da sua primeira requisição ao provedor até um conversor pronto para produção com um cliente HTTP compartilhado, leituras em cache, dinheiro seguro com decimais e tratamento de erros tipado.

Ao final você terá um serviço FastAPI pequeno e autônomo, apoiado pela documentação da API da Finexly e sua cobertura de mais de 170 moedas. Se você já leu nosso tutorial da API de moedas em Python, este é o complemento em nível de serviço que mostra como as peças se encaixam em um backend ASGI real. Enquanto o tutorial de Django se apoia no ORM e na camada de templates, o FastAPI mantém tudo enxuto e orientado a async.

Por que o FastAPI é uma ótima escolha para um microsserviço de moedas

A conversão de moedas parece trivial — multiplicar um valor por uma taxa — mas fazê-la bem como serviço envolve preocupações para as quais o FastAPI foi projetado:

I/O assíncrono real. Buscar taxas depende da rede. Com endpoints async def e um cliente HTTP assíncrono, um único worker pode atender muitas conversões simultâneas enquanto as requisições estão em andamento, em vez de bloquear uma thread por chamada.
Contratos tipados. Modelos Pydantic validam os parâmetros de consulta recebidos e formatam o JSON de saída, então um amount malformado ou um código de moeda desconhecido é rejeitado com um 422 claro antes de chegar à sua lógica.
Documentação interativa gratuita. Cada endpoint que você escreve aparece no Swagger UI em /docs, o que torna o serviço autodocumentado para o resto da sua equipe.
Gerenciamento de lifespan. O FastAPI oferece um lugar limpo para abrir um único pool de conexões HTTP na inicialização e drená-lo no desligamento — crítico para o desempenho, como você verá abaixo.

A arquitetura que construímos é um proxy fino e consciente do cache: seu serviço chama a API de taxas, faz cache da resposta por uma janela curta e serve conversões ao seu próprio frontend ou outros microsserviços. Esse padrão mantém suas chaves de API no servidor, isola você dos limites de taxa do provedor e permite adicionar regras de negócio (arredondamento, margens, listas de permissão) em um único lugar.

Pré-requisitos e configuração do projeto

Você vai precisar de Python 3.11 ou superior. Crie um ambiente virtual e instale as dependências:

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

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

fastapi[standard] traz o Uvicorn (o servidor ASGI) e a CLI. httpx é nosso cliente HTTP assíncrono, e pydantic-settings cuida da configuração a partir de variáveis de ambiente.

Pegue uma chave de API gratuita no seu painel da Finexly e exporte-a para que nunca vá parar no controle de versão:

export FINEXLY_API_KEY="your_api_key_here"

Crie a estrutura do projeto:

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

Passo 1: Configuração com pydantic-settings

Mantenha a configuração em um único lugar tipado. pydantic-settings lê variáveis de ambiente e um arquivo .env, e falha ruidosamente se um valor obrigatório estiver ausente.

# 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

Como o objeto de configuração é criado em tempo de importação, uma chave ausente impede que o serviço inicialize em vez de falhar na primeira requisição — exatamente o que você quer em um pipeline de deploy.

Passo 2: Um cliente HTTP assíncrono compartilhado via lifespan

Esta é a decisão de desempenho mais importante de todo o serviço. Não crie um novo httpx.AsyncClient por requisição. Cada cliente possui um pool de conexões; criar um por chamada descarta as conexões keep-alive e força um novo handshake TLS a cada vez. Em vez disso, abra um cliente quando a app inicia e reutilize-o por toda a vida do processo.

A forma moderna de fazer isso é o gerenciador de contexto lifespan do FastAPI. (Os antigos decoradores @app.on_event("startup") e @app.on_event("shutdown") foram descontinuados no FastAPI 0.93+ em favor do lifespan.) Usar async with garante que o pool de conexões seja drenado e os sockets fechados no desligamento, mesmo se a inicialização for interrompida.

# 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

Agora cada endpoint pega emprestado o mesmo cliente do pool por meio de uma dependência, e ninguém precisa pensar em abrir ou fechar conexões.

Passo 3: Modelos tipados de requisição e resposta

Modelos Pydantic dão a você validação e respostas autodocumentadas. Observe o uso de Decimal para valores monetários: floats binários não conseguem representar frações decimais com exatidão, e erros de arredondamento em dinheiro são um bug real, não uma curiosidade.

# 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

O FastAPI renderizará esses modelos no schema OpenAPI e fará coerção/validação dos valores automaticamente, então uma requisição com um valor sem sentido é rejeitada antes de seu handler rodar.

Passo 4: Buscar taxas com um pequeno cache TTL

As taxas não mudam a cada milissegundo, e martelar a API do provedor a cada requisição é lento e uma forma rápida de esgotar sua cota. Um pequeno cache em memória com tempo de vida (TTL) suaviza isso. Para um serviço de processo único, este cache baseado em dicionário é suficiente; para múltiplos workers, troque por Redis depois sem mexer nos pontos de chamada.

# 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

Repare como as falhas do provedor são traduzidas em códigos de status HTTP significativos para os seus clientes. Um 429 do provedor vira um 503 com uma dica de nova tentativa; uma falha de autenticação vira um 502 para você nunca vazar detalhes de credenciais rio abaixo. Nosso guia complementar sobre cache e tratamento de erros em APIs de moedas aprofunda esses padrões.

Passo 5: Os endpoints /convert e /rates

Agora os endpoints em si são minúsculos, porque todo o trabalho pesado vive em fetch_rates. Validamos os parâmetros de consulta com o Query do FastAPI, fazemos matemática segura com decimais e retornamos modelos tipados.

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

Execute localmente:

fastapi dev app/main.py

Depois abra http://127.0.0.1:8000/docs para o Swagger UI interativo, ou acesse o endpoint diretamente:

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

Você agora tem um microsserviço de moedas funcionando. Se preferir não rodar sua própria matemática de conversão, a Finexly também oferece um conversor de moedas hospedado e um endpoint de conversão direto que você pode chamar.

Passo 6: Uma nota sobre precisão decimal e moedas sem casas decimais

Duas armadilhas do dinheiro pegam quase todos os serviços de moedas:

Nunca use floats binários para valores. 0.1 + 0.2 não é 0.3 na aritmética de ponto flutuante. Analise taxas e valores por meio de Decimal(str(value)) e quantize o resultado final, como fizemos acima.
Nem toda moeda tem duas casas decimais. A ISO 4217 define unidades menores por moeda: JPY e KRW têm zero casas decimais, enquanto BHD e KWD têm três. Fixar no código .quantize(Decimal("0.01")) vai arredondar errado o iene e o dinar silenciosamente. Um serviço de produção deve buscar o expoente correto por moeda e quantizar de acordo.

Acertar isso é a diferença entre uma demo e algo que você pode colocar diante de transações reais.

Passo 7: Testando o serviço

Como o cliente HTTP é injetado como dependência, os testes são fáceis — sobrescreva a dependência com um mock e você nunca toca na rede:

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

O TestClient executa os eventos de lifespan por você, então o cliente compartilhado é configurado e desmontado exatamente como em produção.

Rumo à produção: escalabilidade e limites de taxa

Algumas coisas para planejar antes de publicar:

Múltiplos workers. Sob Uvicorn/Gunicorn com vários workers, seu cache em memória é por processo. Migre para Redis para um cache compartilhado, de modo que uma taxa buscada por um worker sirva a todos. A separação em fetch_rates torna isso uma mudança de uma única função.
Cota do provedor. O cache é sua primeira linha de defesa. Com um TTL de 5 minutos, uma moeda base custa no máximo 12 chamadas ao provedor por hora, independentemente de quantas conversões você sirva. Revise os limites do seu plano na página de preços e ajuste o TTL ao seu nível.
Taxas históricas. Para faturas, relatórios ou trilhas de auditoria você vai querer um endpoint /historical que consulte taxas de uma data específica. Os mesmos padrões de cliente e cache se aplicam; apenas indexe o cache por (base, date).
Observabilidade. Registre as taxas de acerto/erro do cache e a latência do provedor para ajustar o TTL com dados reais.

Se você ainda está escolhendo um provedor, nossa página para comparar APIs de moedas apresenta cobertura, limites e preços lado a lado.

Perguntas frequentes

Por que usar httpx em vez de requests no FastAPI?

requests é síncrono e bloqueia o loop de eventos, o que anula o propósito de um framework assíncrono. httpx oferece um cliente totalmente assíncrono (httpx.AsyncClient) com pool de conexões, timeouts e transportes com retentativas, então ele se integra diretamente aos endpoints async def do FastAPI sem bloquear outras requisições.

Devo criar o cliente httpx por requisição ou uma vez só?

Uma vez só. Crie um único httpx.AsyncClient no gerenciador de contexto lifespan e reutilize-o em todas as requisições. Um cliente por requisição descarta o pool de conexões e força um novo handshake TLS a cada chamada, o que é mensuravelmente mais lento sob carga e pode esgotar os sockets.

Como evito atingir o limite de taxa da API de moedas?

Faça cache das respostas do provedor com um TTL curto (por exemplo, 5 minutos). Como as taxas se movem devagar, o cache por moeda base reduz milhares de conversões de usuários a um punhado de chamadas ao provedor por hora. Para deploys com vários workers, use Redis para que o cache seja compartilhado entre processos.

Como lido com moedas com números diferentes de casas decimais?

Use o tipo Decimal do Python, nunca float, e quantize o resultado ao número correto de unidades menores de cada moeda conforme a ISO 4217. JPY e KRW usam zero casas decimais, a maioria usa duas, e BHD/KWD usam três. Busque o expoente por moeda em vez de fixar duas casas decimais no código.

Posso usar esse padrão para um framework síncrono como Flask ou Django?

Os padrões de cache e Decimal se transferem diretamente, mas as partes de cliente compartilhado e endpoint assíncrono são específicas do FastAPI. Para Django, veja nosso tutorial dedicado da API de moedas em Django, que usa o backend de cache e o ORM do framework.

Comece a construir

Você agora tem um microsserviço de moedas completo, assíncrono e consciente do cache em FastAPI — tipado com Pydantic, seguro com Decimal e resiliente a falhas do provedor. É pequeno o bastante para ler de uma vez e estruturado o bastante para crescer até um serviço de produção real.

Pronto para integrar taxas de câmbio em tempo real ao seu projeto? Obtenha sua chave gratuita da API da Finexly — sem cartão de crédito. Comece com um plano gratuito generoso cobrindo mais de 170 moedas e escale conforme seu tráfego cresce. Prefere explorar primeiro? Veja o resumo da API de moedas gratuita ou mergulhe na documentação da API.