Receta: FastAPI#

Un proyecto completo con FastAPI y el SDK apipay, usando el cliente asíncrono y su lifespan para que el pool de conexiones se cierre bien. Cinco archivos.

mi-tienda/
├── requirements.txt
├── .env.example
└── app/
    ├── __init__.py
    ├── config.py     variables de entorno, validadas al arrancar
    ├── store.py      persistencia de juguete (en produccion, tu base de datos)
    ├── pagina.py     el HTML del checkout, sin un solo campo de tarjeta
    └── main.py       las tres rutas

`requirements.txt`#

apipay>=1.0.0,<2
fastapi>=0.115
uvicorn[standard]>=0.34

httpx llega como dependencia de apipay: es la única que el SDK necesita en runtime, y le da cliente síncrono y asíncrono con la misma superficie.

`.env.example`#

# Backoffice > API keys. La sk_ jamas llega al navegador.
APIPAY_SECRET_KEY=sk_test_EJEMPLO000000000000000000000000
APIPAY_PUBLIC_KEY=pk_test_EJEMPLO000000000000000000000000

# Backoffice > Webhooks. Se muestra UNA sola vez al crear el endpoint.
APIPAY_WEBHOOK_SECRET=whsec_test_EJEMPLO0000000000000000000

# URL publica de esta app: la usan return_url y el endpoint de webhooks.
PUBLIC_BASE_URL=http://127.0.0.1:8000

`app/config.py`#

"""Configuracion leida del entorno y validada al importar: nada de fallos tardios."""

from __future__ import annotations

import os
from dataclasses import dataclass


def _requerido(nombre: str) -> str:
    valor = os.environ.get(nombre, "")
    if not valor:
        raise RuntimeError(f"Falta la variable de entorno {nombre}")
    return valor


@dataclass(frozen=True, slots=True)
class Config:
    secret_key: str
    public_key: str
    webhook_secret: str
    public_base_url: str


CONFIG = Config(
    secret_key=_requerido("APIPAY_SECRET_KEY"),
    public_key=_requerido("APIPAY_PUBLIC_KEY"),
    webhook_secret=_requerido("APIPAY_WEBHOOK_SECRET"),
    public_base_url=_requerido("PUBLIC_BASE_URL"),
)

`app/store.py`#

"""Persistencia de juguete.

En produccion cada funcion de aqui es una consulta a tu base de datos, y
``registrar_evento`` es un INSERT con UNIQUE(event_id) dentro de la MISMA
transaccion que el efecto de negocio: los reintentos de entrega pueden llegar a
instancias distintas del proceso.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Literal

Estado = Literal["pendiente", "pagada", "fallida", "reembolsada"]


@dataclass(slots=True)
class Orden:
    referencia: str
    amount_minor: int
    currency: str
    estado: Estado = "pendiente"
    intent_id: str | None = None
    client_secret: str | None = None
    idempotency_key: str | None = None


_ordenes: dict[str, Orden] = {}
_eventos: set[str] = set()


def obtener_o_crear(referencia: str, amount_minor: int, currency: str) -> Orden:
    orden = _ordenes.get(referencia)
    if orden is None:
        orden = Orden(referencia=referencia, amount_minor=amount_minor, currency=currency)
        _ordenes[referencia] = orden
    return orden


def por_intent(intent_id: str) -> Orden | None:
    return next((o for o in _ordenes.values() if o.intent_id == intent_id), None)


def registrar_evento(event_id: str) -> bool:
    """True la primera vez que se ve el evento; False en los reintentos."""
    if event_id in _eventos:
        return False
    _eventos.add(event_id)
    return True

`app/pagina.py`#

"""HTML del checkout.

PCI DSS SAQ A: aqui no hay ningun input de tarjeta, ni CVV, ni autocomplete de
tarjeta. La captura ocurre dentro del iframe alojado de la pasarela.
"""

from __future__ import annotations

import json

from .store import Orden


def pagina_checkout(orden: Orden, public_key: str) -> str:
    total = (
        f"{orden.amount_minor:,}".replace(",", ".")
        if orden.currency == "CLP"
        else f"{orden.amount_minor / 100:,.2f}"
    )
    return f"""<!doctype html>
<html lang="es-CL">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Pagar orden {orden.referencia}</title>
  </head>
  <body>
    <h1>Orden {orden.referencia}</h1>
    <p>Total: {total} {orden.currency}</p>
    <p id="apipay-error" role="alert"></p>
    <div id="apipay-checkout"></div>

    <script src="https://checkout.apipay.io/sdk/v1.0.0/apipay.js" crossorigin="anonymous" defer></script>
    <script defer>
      addEventListener('DOMContentLoaded', function () {{
        var errorBox = document.querySelector('#apipay-error');
        ApiPay.init({{ publicKey: {json.dumps(public_key)} }}).checkout({{
          clientSecret: {json.dumps(orden.client_secret)},
          container: '#apipay-checkout',
          locale: 'es-CL',
          onSuccess: function (r) {{
            location.assign('/gracias?pi=' + encodeURIComponent(r.paymentIntentId));
          }},
          onError: function (e) {{
            // Decidir SIEMPRE por code, nunca por el texto del mensaje.
            errorBox.textContent = (e.code === 'intent_expired' || e.code === 'session_invalid')
              ? 'La sesion de pago expiro. Recarga la pagina.'
              : 'No pudimos procesar el pago. Prueba con otro medio.';
          }},
          onCancel: function () {{ location.assign('/carro'); }},
        }});
      }});
    </script>
  </body>
</html>"""

`app/main.py`#

"""Tres rutas: crear el intent y servir el widget, la pagina de gracias, y el webhook."""

from __future__ import annotations

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

from apipay import (
    APIConnectionError,
    APIError,
    AsyncApiPay,
    ConfigurationError,
    InvalidPayloadError,
    SignatureVerificationError,
    construct_event,
)
from fastapi import FastAPI, Header, Request, Response
from fastapi.responses import HTMLResponse

from .config import CONFIG
from .pagina import pagina_checkout
from .store import Estado, obtener_o_crear, por_intent, registrar_evento

# Terminado en 00: la pasarela sandbox aprueba. Prueba con 05, 13 y 42.
AMOUNT_MINOR = 1_499_000
CURRENCY = "CLP"


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Un cliente por proceso. Una clave pk_ se rechazaria aqui, al construir.
    app.state.apipay = AsyncApiPay(CONFIG.secret_key, timeout=30.0, max_retries=2)
    try:
        yield
    finally:
        await app.state.apipay.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/checkout/{referencia}", response_class=HTMLResponse)
async def checkout(referencia: str, request: Request) -> HTMLResponse:
    """Crea el intent si hace falta y sirve la pagina del widget.

    Esta misma ruta es la pagina de RETORNO de las pasarelas con redireccion: se
    sirve otra vez con el mismo client_secret y el widget dispara el segundo
    confirm con el token_ws que copio de la URL.
    """
    client: AsyncApiPay = request.app.state.apipay
    orden = obtener_o_crear(referencia, AMOUNT_MINOR, CURRENCY)

    if orden.client_secret is None:
        # Clave estable derivada de la orden: si hay que reintentar, se reintenta con
        # ESTA, nunca con una nueva. El SDK no reintenta POST por diseno.
        clave = f"orden-{orden.referencia}-cobro"
        cuerpo = {
            "amount_minor": orden.amount_minor,
            "currency": orden.currency,
            "gateway_id": "sandbox",
            "description": f"Orden {orden.referencia}",
            "customer_email": "cliente@example.com",
            "return_url": f"{CONFIG.public_base_url}/checkout/{orden.referencia}",
            "metadata": {"order_id": orden.referencia},
        }
        try:
            creado = await client.payment_intents.create(cuerpo, idempotency_key=clave)
        except APIConnectionError:
            # No consta si llego. Reintentar con la MISMA clave es seguro: la API
            # devuelve la respuesta original en vez de crear un segundo cobro.
            creado = await client.payment_intents.create(cuerpo, idempotency_key=clave)
        except APIError as error:
            return HTMLResponse(
                f"<h1>No pudimos iniciar el pago</h1><p>code: {error.code}</p>",
                status_code=502,
            )

        intent = creado.resource
        orden.intent_id = intent["id"]
        orden.client_secret = intent.get("client_secret")
        orden.idempotency_key = creado.idempotency_key

    return HTMLResponse(pagina_checkout(orden, CONFIG.public_key))


@app.get("/gracias", response_class=HTMLResponse)
async def gracias(pi: str, request: Request) -> HTMLResponse:
    client: AsyncApiPay = request.app.state.apipay
    # Un GET si se reintenta solo ante 429 y 5xx, con backoff y jitter completo.
    intent = await client.payment_intents.retrieve(pi)
    orden = por_intent(intent["id"])
    estado_orden = orden.estado if orden is not None else "desconocida"
    return HTMLResponse(
        f"<h1>Gracias</h1><p>Pago <code>{intent['id']}</code> en estado "
        f"<strong>{intent['status']}</strong>.</p>"
        f"<p>Orden: <strong>{estado_orden}</strong> (la confirma el webhook).</p>"
    )


@app.post("/webhooks/apipay")
async def webhook(
    request: Request,
    apipay_signature: str = Header(default="", alias="ApiPay-Signature"),
) -> Response:
    # Bytes CRUDOS, antes de cualquier parseo. Nada de request.json() aqui.
    raw = await request.body()

    try:
        # Verifica t= y v1= (HMAC-SHA256 sobre "{t}." + bytes, tolerancia 300 s,
        # varios v1 durante la rotacion, comparacion en tiempo constante).
        event = construct_event(raw, apipay_signature, CONFIG.webhook_secret)
    except SignatureVerificationError:
        return Response(status_code=400)
    except InvalidPayloadError:
        return Response(status_code=400)
    except ConfigurationError:
        # Falta el secreto en el entorno: fallo de despliegue, no del emisor.
        return Response(status_code=500)

    # Deduplicar por evt_ ANTES de tocar la orden: la entrega es at-least-once y
    # la plataforma reintenta hasta cinco veces (1m, 5m, 30m, 2h, 12h).
    if not registrar_evento(event["id"]):
        return Response(status_code=200)

    recurso: dict[str, Any] = event["data"]["object"]
    tipo = event["type"]

    if tipo == "payment_intent.succeeded":
        _marcar(recurso.get("id", ""), "pagada")
    elif tipo in {
        "payment_intent.failed",
        "payment_intent.expired",
        "payment_intent.canceled",
    }:
        _marcar(recurso.get("id", ""), "fallida")
    elif tipo == "refund.succeeded":
        _marcar(recurso.get("payment_intent_id", ""), "reembolsada")
    # Los tipos nuevos del catalogo son un cambio ADITIVO: ignorar sin fallar.

    # 2xx rapido. Emails y facturacion van a una cola.
    return Response(status_code=200)


def _marcar(intent_id: str, estado: Estado) -> None:
    if not intent_id:
        return
    orden = por_intent(intent_id)
    if orden is not None:
        orden.estado = estado

Arrancar#

python -m venv .venv && . .venv/bin/activate     # Windows: .venv\Scripts\activate
pip install -r requirements.txt
cp .env.example .env                              # y rellena las claves de modo test

set -a && . ./.env && set +a                      # carga el .env en el entorno
uvicorn app.main:app --reload --port 8000

# En otra terminal, expon el puerto para que la plataforma alcance tu webhook:
# cloudflared tunnel --url http://127.0.0.1:8000
# ...y registra <la-url-publica>/webhooks/apipay en el backoffice.

Abre http://127.0.0.1:8000/checkout/4831 y paga.

Probar los cuatro desenlaces#

La pasarela sandbox decide por los dos últimos dígitos de amount_minor. Cambia AMOUNT_MINOR y usa otra referencia de orden:

`AMOUNT_MINOR`	Qué pasa	Estado final
`1_499_000` (…`00`)	Aprueba de inmediato	SUCCEEDED
`1_499_005` (…`05`)	`402 card_declined` en el confirm	FAILED
`1_499_013` (…`13`)	Timeout; resuelve por webhook a los 60 s	FAILED
`1_499_042` (…`42`)	`REQUIRES_ACTION` con `redirect_url`	SUCCEEDED o FAILED

Conciliación con el iterador de cursor#

Un cron de conciliación no necesita tocar cursor ni next_cursor jamás:

from datetime import UTC, datetime, timedelta

from apipay import ApiPay  # el cliente sincrono va bien para un job

desde = (datetime.now(UTC) - timedelta(days=1)).isoformat()

with ApiPay(CONFIG.secret_key) as client:
    pager = client.transactions.list({"status": "APPROVED", "from": desde, "limit": 100})
    for txn in pager:            # pide la pagina siguiente cuando hace falta
        conciliar(txn["id"], txn["amount_minor"], txn["currency"])

Errores comunes en este stack#

Síntoma	Causa
La firma nunca cuadra	Se usó `await request.json()` o un modelo Pydantic en el endpoint
La firma nunca cuadra, y se usa `body()`	Un proxy inverso reescribe el cuerpo
`SignatureVerificationError` sólo en producción	Reloj del servidor desfasado más de 300 s
`RuntimeError: Falta la variable...` al arrancar	El `.env` no se cargó en el entorno del proceso
La orden se marca pagada dos veces	Falta la deduplicación por `event["id"]`
`ConfigurationError` al construir el cliente	Se puso la `pk_` en `APIPAY_SECRET_KEY`

Siguientes pasos#

Webhooks — rotación de secreto, reintentos y deduplicación.
Referencia de SDKs — el cliente asíncrono y la política de reintentos.
Modo test — la tabla completa de sandbox.

Receta: FastAPI#

requirements.txt#

.env.example#

app/config.py#

app/store.py#

app/pagina.py#

app/main.py#