Cómo crear una API de tipos de cambio con FastAPI (httpx asíncrono, caché y Pydantic)

Si estás creando un backend fintech, un checkout multidivisa o un servicio interno de precios en Python, tarde o temprano necesitarás tu propio endpoint de tipos de cambio. Crear una API de tipos de cambio con FastAPI te da un microservicio rápido, asíncrono y totalmente tipado que envuelve a un proveedor de tasas en vivo, añade caché para mantenerte dentro de un plan gratuito y expone un endpoint /convert limpio que tus otros servicios pueden llamar. FastAPI encaja de forma natural aquí: está construido sobre ASGI para I/O asíncrono real, valida solicitudes y respuestas con Pydantic y genera documentación interactiva OpenAPI de forma gratuita. Este tutorial recorre todo el camino: desde tu primera solicitud al proveedor hasta un conversor listo para producción con un cliente HTTP compartido, lecturas en caché, dinero seguro con decimales y manejo de errores tipado.

Al final tendrás un servicio FastAPI pequeño y autónomo respaldado por la documentación de la API de Finexly y su cobertura de más de 170 monedas. Si ya has leído nuestro tutorial de la API de monedas en Python, este es el complemento a nivel de servicio que muestra cómo encajan las piezas en un backend ASGI real. Mientras que el tutorial de Django se apoya en el ORM y la capa de plantillas, FastAPI mantiene todo ligero y orientado a async.

Por qué FastAPI encaja bien para un microservicio de monedas

La conversión de monedas parece trivial —multiplicar un importe por una tasa— pero hacerlo bien como servicio toca aspectos para los que FastAPI fue diseñado:

I/O asíncrono real. Obtener tasas depende de la red. Con endpoints async def y un cliente HTTP asíncrono, un solo worker puede atender muchas conversiones concurrentes mientras las solicitudes están en curso, en lugar de bloquear un hilo por llamada.
Contratos tipados. Los modelos Pydantic validan los parámetros de consulta entrantes y dan forma al JSON saliente, de modo que un amount mal formado o un código de moneda desconocido se rechaza con un 422 claro antes de llegar a tu lógica.
Documentación interactiva gratuita. Cada endpoint que escribes aparece en Swagger UI en /docs, lo que hace que el servicio se documente solo para el resto de tu equipo.
Gestión del lifespan. FastAPI te da un lugar limpio para abrir un único pool de conexiones HTTP al arrancar y drenarlo al apagar, algo crítico para el rendimiento, como verás más abajo.

La arquitectura que construimos es un proxy fino y consciente de la caché: tu servicio llama a la API de tasas, cachea la respuesta durante una ventana corta y sirve conversiones a tu propio frontend u otros microservicios. Ese patrón mantiene tus claves de API en el servidor, te aísla de los límites de tasa del proveedor y te permite añadir reglas de negocio (redondeo, márgenes, listas permitidas) en un solo lugar.

Requisitos previos y configuración del proyecto

Necesitarás Python 3.11 o superior. Crea un entorno virtual e instala las dependencias:

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

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

fastapi[standard] incluye Uvicorn (el servidor ASGI) y la CLI. httpx es nuestro cliente HTTP asíncrono y pydantic-settings gestiona la configuración a partir de variables de entorno.

Consigue una clave de API gratuita en tu panel de Finexly y expórtala para que nunca llegue al control de versiones:

export FINEXLY_API_KEY="your_api_key_here"

Crea la estructura del proyecto:

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

Paso 1: Configuración con pydantic-settings

Mantén la configuración en un único lugar tipado. pydantic-settings lee variables de entorno y un archivo .env, y falla ruidosamente si falta un valor requerido.

# 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 el objeto de configuración se crea en tiempo de importación, una clave ausente impide que el servicio arranque en lugar de fallar en la primera solicitud, justo lo que quieres en un pipeline de despliegue.

Paso 2: Un cliente HTTP asíncrono compartido mediante lifespan

Esta es la decisión de rendimiento más importante de todo el servicio. No crees un nuevo httpx.AsyncClient por cada solicitud. Cada cliente posee un pool de conexiones; crear uno por llamada descarta las conexiones keep-alive y fuerza un nuevo handshake TLS cada vez. En su lugar, abre un cliente al arrancar la app y reutilízalo durante toda la vida del proceso.

La forma moderna de hacerlo es el gestor de contexto lifespan de FastAPI. (Los antiguos decoradores @app.on_event("startup") y @app.on_event("shutdown") quedaron obsoletos en FastAPI 0.93+ a favor de lifespan.) Usar async with garantiza que el pool de conexiones se drene y los sockets se cierren al apagar, incluso si el arranque se interrumpe.

# 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

Ahora cada endpoint toma prestado el mismo cliente del pool a través de una dependencia, y nadie tiene que pensar en abrir o cerrar conexiones.

Paso 3: Modelos tipados de solicitud y respuesta

Los modelos Pydantic te dan validación y respuestas autodocumentadas. Fíjate en el uso de Decimal para los importes monetarios: los flotantes binarios no pueden representar fracciones decimales con exactitud, y los errores de redondeo en dinero son un bug real, no una curiosidad.

# 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 renderizará estos modelos en el esquema OpenAPI y coaccionará/validará los valores automáticamente, de modo que una solicitud con un importe sin sentido se rechaza antes de que se ejecute tu manejador.

Paso 4: Obtener tasas con una pequeña caché TTL

Las tasas no cambian milisegundo a milisegundo, y golpear la API del proveedor en cada solicitud es lento y una forma rápida de agotar tu cuota. Una pequeña caché en memoria con tiempo de vida (TTL) suaviza esto. Para un servicio de un solo proceso, esta caché basada en diccionario es suficiente; para varios workers, cámbiala por Redis más tarde sin tocar los sitios de llamada.

# 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

Fíjate en cómo los fallos del proveedor se traducen en códigos de estado HTTP significativos para tus clientes. Un 429 del proveedor se convierte en un 503 con una pista de reintento; un fallo de autenticación se convierte en un 502 para que nunca filtres detalles de credenciales aguas abajo. Nuestra guía complementaria sobre caché y manejo de errores en APIs de monedas profundiza en estos patrones.

Paso 5: Los endpoints /convert y /rates

Ahora los endpoints en sí son diminutos, porque todo el trabajo duro vive en fetch_rates. Validamos los parámetros de consulta con Query de FastAPI, hacemos matemática segura con decimales y devolvemos 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"}

Ejecútalo localmente:

fastapi dev app/main.py

Luego abre http://127.0.0.1:8000/docs para la Swagger UI interactiva, o llama al endpoint directamente:

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

Ya tienes un microservicio de monedas funcionando. Si prefieres no ejecutar tu propia matemática de conversión, Finexly también ofrece un conversor de monedas alojado y un endpoint de conversión directo que puedes llamar.

Paso 6: Nota sobre precisión decimal y monedas sin decimales

Dos trampas del dinero atrapan a casi todos los servicios de monedas:

Nunca uses flotantes binarios para importes. 0.1 + 0.2 no es 0.3 en aritmética de punto flotante. Analiza tasas e importes con Decimal(str(value)) y cuantiza el resultado final, como hicimos arriba.
No toda moneda tiene dos decimales. ISO 4217 define unidades menores por moneda: JPY y KRW tienen cero decimales, mientras que BHD y KWD tienen tres. Codificar a mano .quantize(Decimal("0.01")) redondeará mal el yen y el dinar de forma silenciosa. Un servicio de producción debería buscar el exponente correcto por moneda y cuantizar en consecuencia.

Hacer esto bien es la diferencia entre una demo y algo que puedes poner frente a transacciones reales.

Paso 7: Probar el servicio

Como el cliente HTTP se inyecta como dependencia, las pruebas son fáciles: sobrescribe la dependencia con un mock y nunca tocas la red:

# 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 ejecuta los eventos de lifespan por ti, así que el cliente compartido se configura y se desmonta exactamente como en producción.

Rumbo a producción: escalado y límites de tasa

Algunas cosas que planificar antes de desplegar:

Múltiples workers. Bajo Uvicorn/Gunicorn con varios workers, tu caché en memoria es por proceso. Muévete a Redis para una caché compartida, de modo que una tasa obtenida por un worker sirva a todos. La separación en fetch_rates hace de esto un cambio de una sola función.
Cuota del proveedor. La caché es tu primera línea de defensa. Con un TTL de 5 minutos, una moneda base cuesta como máximo 12 llamadas al proveedor por hora, sin importar cuántas conversiones sirvas. Revisa los límites de tu plan en la página de precios y ajusta el TTL a tu nivel.
Tasas históricas. Para facturas, informes o pistas de auditoría querrás un endpoint /historical que consulte tasas de una fecha concreta. Se aplican los mismos patrones de cliente y caché; solo indexa la caché por (base, date).
Observabilidad. Registra las proporciones de acierto/fallo de caché y la latencia del proveedor para ajustar el TTL con datos reales.

Si aún estás eligiendo proveedor, nuestra página para comparar APIs de monedas muestra cobertura, límites y precios en paralelo.

Preguntas frecuentes

¿Por qué usar httpx en lugar de requests en FastAPI?

requests es síncrono y bloquea el bucle de eventos, lo que anula el propósito de un framework asíncrono. httpx ofrece un cliente totalmente asíncrono (httpx.AsyncClient) con pool de conexiones, timeouts y transportes con reintentos, así que se integra directamente en los endpoints async def de FastAPI sin bloquear otras solicitudes.

¿Debo crear el cliente httpx por solicitud o una sola vez?

Una sola vez. Crea un único httpx.AsyncClient en el gestor de contexto lifespan y reutilízalo en todas las solicitudes. Un cliente por solicitud descarta el pool de conexiones y fuerza un nuevo handshake TLS en cada llamada, lo que es mediblemente más lento bajo carga y puede agotar los sockets.

¿Cómo evito alcanzar el límite de tasa de la API de monedas?

Cachea las respuestas del proveedor con un TTL corto (por ejemplo, 5 minutos). Como las tasas se mueven despacio, cachear por moneda base reduce miles de conversiones de usuarios a un puñado de llamadas al proveedor por hora. Para despliegues con varios workers, usa Redis para que la caché sea compartida entre procesos.

¿Cómo manejo monedas con distinto número de decimales?

Usa el tipo Decimal de Python, nunca float, y cuantiza el resultado al número correcto de unidades menores de cada moneda según ISO 4217. JPY y KRW usan cero decimales, la mayoría usa dos, y BHD/KWD usan tres. Busca el exponente por moneda en lugar de codificar a mano dos decimales.

¿Puedo usar este patrón para un framework síncrono como Flask o Django?

Los patrones de caché y Decimal se transfieren directamente, pero las partes de cliente compartido y endpoint asíncrono son específicas de FastAPI. Para Django, consulta nuestro tutorial dedicado de la API de monedas en Django, que usa el backend de caché y el ORM del framework.

Empieza a construir

Ahora tienes un microservicio de monedas completo, asíncrono y consciente de la caché en FastAPI: tipado con Pydantic, seguro con Decimal y resistente a fallos del proveedor. Es lo bastante pequeño para leerlo de una sentada y lo bastante estructurado para crecer hasta un servicio de producción real.

¿Listo para integrar tipos de cambio en tiempo real en tu proyecto? Consigue tu clave gratuita de la API de Finexly: sin tarjeta de crédito. Empieza con un generoso plan gratuito que cubre más de 170 monedas y escala a medida que crece tu tráfico. ¿Prefieres explorar primero? Echa un vistazo al resumen de la API de monedas gratuita o sumérgete en la documentación de la API.