Ir para o conteúdo

Protocolo de Pix: um contrato, vários provedores

Seu serviço cobra por Pix. Hoje o provedor é o OpenPix; amanhã pode ser o Mercado Pago, porque a taxa mudou ou porque o cliente já tem conta lá.

Se o seu service fala a língua do provedor, essa troca reescreve o service. Esta receita mostra como não falar.

O problema, concretamente

Os dois provedores que o SDK já traz discordam em quase tudo que importa:

OpenPix Mercado Pago
valor centavos, num int reais, num float
estados ACTIVE, COMPLETED, EXPIRED 9 estados em Payment, 5 em Order, 4 em OrderTransactionPayment
copia-e-cola brCode qr_code
imagem do QR uma URL Base64
sua referência correlationID external_reference

Escrever if charge.status == "COMPLETED" não acopla seu código ao Pix. Acopla ao OpenPix.

O contrato

from tempest_fastapi_sdk.integrations.payment import (
    PaymentStatus,
    PixCharge,
    PixChargeRequest,
    PixEventType,
    PixPaymentEvent,
    PixPayer,
    PixProvider,
)

Nenhum extra para o contrato

Medido numa instalação sem extra: o import acima e o OpenPixPixProvider resolvem do core. Extra é assunto do cliente HTTP de cada provedor, não do contrato.

PixProvider é um atributo e quatro métodos — é o arquivo integrations/payment/base.py, sem simplificação:

provider_name: str

async def create_pix_charge(self, request: PixChargeRequest) -> PixCharge
async def get_pix_charge(self, charge_id: str) -> PixCharge
async def cancel_pix_charge(self, charge_id: str) -> PixCharge
def parse_webhook(self, event: Any) -> PixPaymentEvent

Protocol, não classe base: o adapter satisfaz por forma, sem herdar nada — a mesma costura que o SDK usa em RateLimitStore, QuotaStore, ModerationBackend e PushDispatcher.

Por que não é runtime_checkable

isinstance contra protocolo runtime-checkable confere só que os nomes existem: um adapter cujo create_pix_charge recebe os argumentos errados passaria na checagem e falharia na cobrança. Quem confere de verdade é tests/integrations/payment/test_contract.py, comparando inspect.signature — e o seu type-checker, se você declarar o tipo como PixProvider (a próxima seção mostra onde).

O que entra: PixChargeRequest

campo tipo para quê
amount_cents int valor em centavos — inteiro, nunca float
reference str seu identificador; volta em PixCharge.reference
description str ou None texto que o pagador vê
expires_in timedelta ou None janela de pagamento
payer PixPayer ou None dados do pagador, quando o provedor aceita

O que o adapter faz com um payer incompleto (v0.270.0)

Todo campo de PixPayer é opcional, mas provedor tem regra própria sobre o conjunto. A OpenPix exige name e pelo menos um de taxID, email ou phone — as três variantes do oneOf dela pedem name mais um contato, então name é necessário nas três e suficiente em nenhuma.

O adapter omite o bloco quando o payer não alcança nenhuma variante, em vez de mandar um bloco que a especificação do provedor não aceita, e em vez de inventar um contato para preenchê-lo — dado inventado acabaria no comprovante do pagador. A cobrança é válida sem bloco de pagador.

reference pode ter caractere reservado

O reference é o identificador que você escolhe, e ele vira o correlationID da OpenPix, que por sua vez entra no path das rotas de leitura e de cancelamento. O SDK escapa o segmento (quote(..., safe="")), então pedido#42 e pedido/1042 endereçam a cobrança certa. Antes da v0.270.0 o # era interpretado como fragmento e a chamada acertava outro recurso — num DELETE.

O que sai: PixCharge

campo tipo para quê
provider str quem emitiu (vem de provider_name)
provider_charge_id str o id que você guarda — é o argumento de get_pix_charge e cancel_pix_charge
reference str o seu identificador, de volta
amount_cents int valor, em centavos
currency str ISO 4217, default BRL
status PaymentStatus o estado sobre o qual você decide
provider_status str o estado como o provedor o nomeia, cru
br_code str ou None copia-e-cola EMV
qr_code_image_url str ou None QR como URL (é o que o OpenPix devolve)
qr_code_base64 str ou None QR como Base64 (é o que o Mercado Pago devolve)
end_to_end_id str ou None identificador da liquidação no Pix
expires_at datetime ou None quando a janela fecha
paid_at datetime ou None quando liquidou
raw dict[str, Any] tudo que o provedor disse além disso

O QR vem nos dois formatos porque os provedores discordam

O contrato carrega qr_code_image_url e qr_code_base64, e preenche o que o provedor entregar. Seu template lê o que existir, em vez de saber qual provedor está atrás.

Cobrando

Um programa inteiro, do cliente HTTP à cobrança:

import asyncio

from tempest_fastapi_sdk import HTTPClient
from tempest_fastapi_sdk.integrations.payment import (
    PaymentStatus,
    PixChargeRequest,
    PixProvider,
)
from tempest_fastapi_sdk.integrations.payment.adapters.openpix import (
    OpenPixPixProvider,
)
from tempest_fastapi_sdk.integrations.payment.openpix import (
    OpenPixClient,
    OpenPixEnvironment,
)


async def cobrar(provider: PixProvider) -> None:
    """Emit a Pix charge and print what the payer needs to see.

    Args:
        provider (PixProvider): Any provider that implements the contract.
    """
    charge = await provider.create_pix_charge(
        PixChargeRequest(
            amount_cents=1990,
            reference="pedido-1042",
            description="Pedido 1042",
        )
    )

    print(charge.status is PaymentStatus.PENDING)
    print(charge.br_code)
    print(charge.provider)


async def main() -> None:
    """Wire the OpenPix client and charge through the contract."""
    http: HTTPClient = HTTPClient(
        base_url=OpenPixEnvironment.SANDBOX.base_url,
        default_headers={"Authorization": "<seu AppID>"},
    )
    await cobrar(OpenPixPixProvider(OpenPixClient(http)))


if __name__ == "__main__":
    asyncio.run(main())

Repare no tipo de cobrar: ele recebe PixProvider, não OpenPixPixProvider. É só isso que separa um service portável de um acoplado.

O valor é int, sempre

amount_cents é um inteiro de centavos. Os dois provedores tipam dinheiro como number na especificação deles — e valor que passou por float é valor que pode estar errado. A conversão para a unidade que cada provedor espera é problema do adapter.

Consultando e cancelando

Criar é um quarto do contrato. O resto do ciclo usa o provider_charge_id que a cobrança devolveu — guarde esse campo junto do seu pedido, porque é o único jeito de voltar ao provedor:

import asyncio

from tempest_fastapi_sdk.integrations.payment import (
    PaymentStatus,
    PixCharge,
    PixChargeRequest,
    PixProvider,
)


async def cobrar_e_acompanhar(provider: PixProvider) -> PixCharge:
    """Create a charge, read it back, and withdraw it if it is still open.

    Args:
        provider (PixProvider): Any provider that implements the contract.

    Returns:
        PixCharge: The charge in its final observed state.
    """
    charge = await provider.create_pix_charge(
        PixChargeRequest(amount_cents=1990, reference="pedido-1042"),
    )
    charge_id: str = charge.provider_charge_id

    current = await provider.get_pix_charge(charge_id)
    if current.status is PaymentStatus.PAID:
        return current

    return await provider.cancel_pix_charge(charge_id)

Consultar não substitui webhook, e webhook não substitui consultar

get_pix_charge é a fonte que você controla: ela responde quando você pergunta. O webhook é a que chega primeiro, e pode não chegar. Um serviço que só escuta webhook fica preso quando a entrega falha; um que só consulta paga latência em cada pedido. A receita OpenPix » mostra os dois lados montados — webhook para reagir, conferência para reconciliar.

Trocando de provedor

O adapter é a única linha que muda. Concentre a escolha num lugar e devolva o contrato, nunca o adapter:

from tempest_fastapi_sdk.integrations.payment import PixProvider
from tempest_fastapi_sdk.integrations.payment.adapters.openpix import (
    OpenPixPixProvider,
)
from tempest_fastapi_sdk.integrations.payment.openpix import OpenPixClient


def build_provider(client: OpenPixClient) -> PixProvider:
    """Choose the Pix provider this deployment charges with.

    Args:
        client (OpenPixClient): The configured OpenPix client.

    Returns:
        PixProvider: The provider, seen through the contract.
    """
    return OpenPixPixProvider(client)

Como o retorno é o Protocol, o type-checker passa a cobrar de você que o resto do service não use nada além do contrato — e é ele quem avisa, não a produção. Num serviço FastAPI, essa função é o corpo do Depends: o router recebe PixProvider e nunca sabe qual adapter chegou — esse é o ponto, e a próxima seção monta o serviço inteiro em volta dele.

Quantos adapters existem hoje: dois

O SDK ships dois adapters prontos: OpenPixPixProvider, em integrations/payment/adapters/openpix.py, e MercadoPagoPixProvider, em integrations/payment/adapters/mercado_pago.py. O Stripe entra por outro caminho, porque não faz Pix. Para trocar de um para o outro sem desligar o primeiro de uma vez, veja Ligando o segundo provedor sem desligar o primeiro.

Na arquitetura do serviço

Até aqui o provider chegou pronto, como argumento. Numa aplicação FastAPI alguém precisa construí-lo — e é essa construção que decide se trocar de provedor é uma linha ou uma refatoração.

Esta seção monta o caminho inteiro, de baixo para cima: cliente HTTP → adapter → dependência → pedido → service → router. O serviço que sai daqui tem dois arquivos que sabem o nome "OpenPix"; todo o resto fala contrato. E ele só libera um pedido depois de reler a cobrança no provedor — o webhook avisa, a API confirma, do mesmo jeito que a receita OpenPix » exige.

Onde cada peça mora

src/
├── core/
│   └── settings.py              # OPENPIX_APP_ID + ambiente + DATABASE_URL
├── api/
│   ├── app.py                   # create_app() + lifespan
│   ├── dependencies/
│   │   ├── orders.py            # sessão por request + o repositório do pedido
│   │   └── payments.py          # HTTPClient -> OpenPixClient -> adapter
│   └── routers/
│       ├── checkout.py          # POST /api/checkout/{order_id}
│       └── webhooks.py          # POST /webhooks/pix (include_in_schema=False)
├── schemas/
│   └── checkout.py              # o que o seu cliente vê
├── services/
│   └── checkout.py              # regra de negócio, escrita só sobre o contrato
└── db/
    ├── models/
    │   └── orders.py            # valor, provider_charge_id e paid_at
    └── repositories/
        └── orders.py            # onde a liberação única (claim_once) mora
Camada Pode importar Nunca importa
api/dependencies HTTPClient, OpenPixClient, o adapter, services —
api/routers as dependências, schemas o adapter, OpenPixClient
services PixProvider, PixCharge, repositories o adapter, fastapi
schemas BaseSchema o contrato e o adapter
db/repositories db/models, claim_once qualquer coisa de pagamento

api/dependencies é a única camada autorizada a conhecer o provedor porque é a única cuja função é montar. É o composition root: o lugar onde o concreto vira contrato, e o único que muda no dia da troca.

Passo 1 — a configuração

from tempest_fastapi_sdk import DatabaseSettings, OpenPixSettings


class Settings(OpenPixSettings, DatabaseSettings):
    """Settings do serviço."""


settings: Settings = Settings()

OpenPixSettings traz OPENPIX_APP_ID e OPENPIX_ENVIRONMENT, e settings.openpix_kwargs() devolve base_url e o header Authorization já resolvidos — os dois argumentos que o HTTPClient precisa. Detalhes dos ambientes em OpenPix ». DatabaseSettings traz DATABASE_URL, e settings.database_kwargs() é o splat do AsyncDatabaseManager do passo 3.

Passo 2 — o cliente HTTP e o provider, montados uma vez

Este é o arquivo inteiro. Ele é o composition root:

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

from fastapi import Depends, FastAPI, Request

from tempest_fastapi_sdk import HTTPClient
from tempest_fastapi_sdk.integrations.payment import PixProvider
from tempest_fastapi_sdk.integrations.payment.adapters import OpenPixPixProvider
from tempest_fastapi_sdk.integrations.payment.openpix import (
    OpenPixClient,
    OpenPixWebhookEvent,
    make_openpix_webhook_dependency,
)

from src.api.dependencies.orders import OrderRepositoryDep
from src.core.settings import settings
from src.services.checkout import CheckoutService


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """Build one HTTP client and one provider for the whole process.

    Args:
        app (FastAPI): The application the provider is stored on.

    Yields:
        None: While the application serves requests.
    """
    http: HTTPClient = HTTPClient(**settings.openpix_kwargs(), timeout=15.0)
    app.state.pix_provider = OpenPixPixProvider(OpenPixClient(http))
    try:
        yield
    finally:
        await http.aclose()


def get_pix_provider(request: Request) -> PixProvider:
    """Hand out the process-wide provider, seen through the contract.

    Args:
        request (Request): The request in flight.

    Returns:
        PixProvider: The configured provider.
    """
    provider: PixProvider = request.app.state.pix_provider
    return provider


PixProviderDep = Annotated[PixProvider, Depends(get_pix_provider)]
"""The contract, injected."""

verified_delivery = make_openpix_webhook_dependency()
"""The provider's own verifier, as a dependency tests can override."""

WebhookDeliveryDep = Annotated[OpenPixWebhookEvent, Depends(verified_delivery)]
"""A delivery whose signature the verifier already accepted."""


def get_checkout_service(
    provider: PixProviderDep,
    orders: OrderRepositoryDep,
) -> CheckoutService:
    """Assemble the service the routers call.

    Args:
        provider (PixProviderDep): The provider, through the contract.
        orders (OrderRepositoryDep): The order repository.

    Returns:
        CheckoutService: The service, ready to charge.
    """
    return CheckoutService(provider, orders)


CheckoutServiceDep = Annotated[CheckoutService, Depends(get_checkout_service)]
"""The checkout service, injected."""

Quatro coisas acontecem aí, e vale ler uma de cada vez.

A montagem são três camadas numa linha. OpenPixPixProvider(OpenPixClient(http)) empilha transporte → cliente do provedor → adapter. Cada uma faz uma coisa: o HTTPClient tem retry, timeout e circuit breaker por host; o OpenPixClient sabe as rotas; o adapter traduz para o contrato.

O HTTPClient é um só, criado no lifespan. Ele é seguro para compartilhar entre requisições do mesmo event loop, e o pool de conexões, o retry e o breaker são estado dele. Criar um por requisição joga as três coisas fora e abre um socket novo em cada checkout — e aclose() no finally é o que fecha o pool no shutdown.

A anotação de retorno é a costura. get_pix_provider promete PixProvider, não OpenPixPixProvider. Daí para cima, o type-checker recusa qualquer uso de algo que não esteja no contrato.

verified_delivery é uma variável nomeada de propósito. Ela é o callable que o Depends registra — e é a chave do app.dependency_overrides no teste. Escrevendo Depends(make_openpix_webhook_dependency()) direto dentro do Annotated, a função fica anônima e não há como substituí-la.

app.state é Any — reanote na leitura

request.app.state.pix_provider não tem tipo: app.state aceita qualquer atributo. Sem a linha provider: PixProvider = ..., o mypy --strict reprova o módulo:

src/api/dependencies/payments.py:50: error: Returning Any from function
declared to return "PixProvider"  [no-any-return]

A anotação não é decoração: é o ponto onde o valor volta a existir para o type-checker. Com ela, o serviço inteiro passa em mypy --strict.

Passo 3 — o pedido, e a coluna que libera uma vez só

O webhook vai chegar mais de uma vez: a OpenPix reentrega quando não recebe 200, e uma entrega capturada continua com assinatura válida para sempre. O pedido precisa de uma coluna que diga "já liberado" — e que só uma requisição consiga preencher.

from datetime import datetime

from sqlalchemy.orm import Mapped, mapped_column

from tempest_fastapi_sdk import BaseModel


class OrderModel(BaseModel):
    """An order, and the Pix charge that pays it."""

    __tablename__ = "orders"

    reference: Mapped[str] = mapped_column(unique=True)
    amount_cents: Mapped[int] = mapped_column()
    provider_charge_id: Mapped[str | None] = mapped_column(default=None)
    paid_at: Mapped[datetime | None] = mapped_column(default=None)

O repositório é quem fala com o banco, então é nele que a liberação única mora:

from uuid import UUID

from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import BaseRepository
from tempest_fastapi_sdk.wallet import claim_once

from src.db.models import OrderModel


class OrderRepository(BaseRepository[OrderModel]):
    """Data access for orders."""

    def __init__(self, session: AsyncSession) -> None:
        """Bind the repository to the request session.

        Args:
            session (AsyncSession): The request-scoped session.
        """
        super().__init__(session, model=OrderModel)

    async def open(
        self, reference: str, amount_cents: int, provider_charge_id: str
    ) -> OrderModel:
        """Persist an order together with the charge that pays it.

        Args:
            reference (str): The order's identifier, sent as the reference.
            amount_cents (int): What the order costs, in cents.
            provider_charge_id (str): How to address the charge later.

        Returns:
            OrderModel: The stored order.
        """
        return await self.add(
            OrderModel(
                reference=reference,
                amount_cents=amount_cents,
                provider_charge_id=provider_charge_id,
            ),
        )

    async def get_by_reference(self, reference: str) -> OrderModel | None:
        """Find an order by the reference its charge carries.

        Args:
            reference (str): The order's identifier.

        Returns:
            OrderModel | None: The order, or None when nothing matches.
        """
        return await self.get_or_none({"reference": reference})

    async def claim_paid(self, order_id: UUID) -> bool:
        """Stamp paid_at, only if no one stamped it before.

        Args:
            order_id (UUID): The order to claim.

        Returns:
            bool: True for the one call that claimed the order.
        """
        return await claim_once(self.session, OrderModel, order_id, "paid_at")

claim_once é um UPDATE orders SET paid_at = now WHERE id = :id AND paid_at IS NULL. Duas entregas do mesmo evento emitem o mesmo UPDATE; só uma encontra a coluna vazia, e a outra recebe False. Ele não faz commit: entra no mesmo bloco de transação que os efeitos da liberação, e uma falha nesses efeitos desfaz a marcação também. É o mesmo guard da receita Carteira ».

E a sessão por request, que o repositório recebe:

from functools import lru_cache
from typing import Annotated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from tempest_fastapi_sdk import AsyncDatabaseManager, session_dependency_for

from src.core.settings import settings
from src.db.repositories import OrderRepository


@lru_cache
def get_db() -> AsyncDatabaseManager:
    """Build the database manager once, on first use.

    Returns:
        AsyncDatabaseManager: The process-wide manager.
    """
    return AsyncDatabaseManager(**settings.database_kwargs())


get_session = session_dependency_for(get_db)
"""One uncommitted session per request; the repository commits."""

SessionDep = Annotated[AsyncSession, Depends(get_session)]
"""The request session, injected."""


def get_order_repository(session: SessionDep) -> OrderRepository:
    """Bind the order repository to the request session.

    Args:
        session (SessionDep): The request session.

    Returns:
        OrderRepository: The repository.
    """
    return OrderRepository(session)


OrderRepositoryDep = Annotated[OrderRepository, Depends(get_order_repository)]
"""The order repository, injected."""

get_session é uma variável nomeada pelo mesmo motivo que verified_delivery: é a chave que o teste do passo 7 troca por um banco temporário.

Passo 4 — o service, que só fala contrato

from datetime import timedelta

from tempest_fastapi_sdk.integrations.payment import (
    PixCharge,
    PixChargeRequest,
    PixEventType,
    PixPaymentEvent,
    PixProvider,
    confirm_pix_payment,
)

from src.db.repositories import OrderRepository


class CheckoutService:
    """Open and settle Pix charges for orders."""

    def __init__(self, provider: PixProvider, orders: OrderRepository) -> None:
        """Take the contract and the repository.

        Args:
            provider (PixProvider): Any provider that implements the contract.
            orders (OrderRepository): Where the order and its charge live.
        """
        self._provider: PixProvider = provider
        self._orders: OrderRepository = orders

    async def open_charge(self, order_id: str, amount_cents: int) -> PixCharge:
        """Charge an order and remember how to address the charge later.

        Args:
            order_id (str): The order's identifier, sent as the reference.
            amount_cents (int): The amount, in cents.

        Returns:
            PixCharge: The created charge, in canonical shape.
        """
        charge = await self._provider.create_pix_charge(
            PixChargeRequest(
                amount_cents=amount_cents,
                reference=order_id,
                description=f"Pedido {order_id}",
                expires_in=timedelta(minutes=30),
            ),
        )
        await self._orders.open(order_id, amount_cents, charge.provider_charge_id)
        return charge

    async def settle(self, event: PixPaymentEvent) -> str | None:
        """Release an order once the provider confirms it is paid.

        The event is only the notice. What authorizes the release is the
        provider answering, on a fresh read of the charge this service
        stored, that it is paid for this order and amount.

        Args:
            event (PixPaymentEvent): The parsed event.

        Returns:
            str | None: The order released by this call, or None when the
            event says something else, the charge is not confirmed, or an
            earlier delivery already released the order.
        """
        if event.type is not PixEventType.CHARGE_PAID or event.charge is None:
            return None
        order = await self._orders.get_by_reference(event.charge.reference)
        if order is None or order.provider_charge_id is None:
            return None
        confirmation = await confirm_pix_payment(
            self._provider,
            order.provider_charge_id,
            reference=order.reference,
            amount_cents=order.amount_cents,
        )
        if not confirmation.paid:
            return None
        async with self._orders.transaction():
            if not await self._orders.claim_paid(order.id):
                return None
        return order.reference

Repare no bloco de imports: nada de adapters, nada de openpix. É a regra que dá para verificar com grep em vez de com revisão.

O settle é onde o dinheiro se move, e cada linha dele está lá por um motivo:

  1. O evento é só o gatilho. Dele sai uma coisa: qual pedido olhar (event.charge.reference). O status e o valor que vieram na entrega não decidem nada.
  2. A releitura usa o id que você guardou. confirm_pix_payment chama provider.get_pix_charge(order.provider_charge_id) — o id gravado no open_charge, não um id que a entrega trouxe. Uma entrega forjada ou reenviada consegue, no máximo, fazer o serviço perguntar ao provedor sobre uma cobrança que já é dele.
  3. Pago é três conferências. A cobrança relida precisa ser deste pedido (reference), estar PAID e ter o valor do pedido (amount_cents, igualdade exata). confirmation.outcome diz qual falhou: REFERENCE_MISMATCH, NOT_PAID ou AMOUNT_MISMATCH.
  4. A liberação é única. claim_paid roda dentro de self._orders.transaction(); a segunda entrega do mesmo evento encontra paid_at preenchido e devolve None. Efeito de liberação — baixar estoque, mandar o e-mail, creditar a carteira do vendedor — entra dentro desse bloco, depois do claim_paid, para cair junto se o efeito falhar.

Erro do provedor não é \"não pago\"

Se o get_pix_charge levanta (timeout, provedor fora), o confirm_pix_payment deixa a exceção subir, e o pedido continua sem liberar — que é o certo para uma pergunta sem resposta. Medido com o get_pix_charge do fake levantando TimeoutError: a entrega responde 500; a reentrega seguinte, com o provedor de volta, responde {"settled": "pedido-1042"}. Quem reentrega é o provedor, quando não recebe 200 (ver OpenPix »).

O reference é o id do seu pedido, e o provider_charge_id é gravado na mesma transação em que a cobrança nasce. Um é como o webhook te encontra; o outro é como você volta ao provedor para consultar ou cancelar. Perder o segundo significa uma cobrança que existe no provedor e que o seu serviço não sabe mais endereçar.

Passo 5 — o router devolve schema seu, não PixCharge

from fastapi import APIRouter, status

from src.api.dependencies import CheckoutServiceDep
from src.schemas import CheckoutCreateSchema, CheckoutResponseSchema

router: APIRouter = APIRouter(prefix="/api/checkout", tags=["checkout"])


@router.post("/{order_id}", status_code=status.HTTP_201_CREATED)
async def open_checkout(
    order_id: str,
    payload: CheckoutCreateSchema,
    service: CheckoutServiceDep,
) -> CheckoutResponseSchema:
    """Open a Pix charge for an order.

    Args:
        order_id (str): The order to charge.
        payload (CheckoutCreateSchema): How much to charge.
        service (CheckoutServiceDep): The checkout service.

    Returns:
        CheckoutResponseSchema: What the payment screen needs.
    """
    charge = await service.open_charge(order_id, payload.amount_cents)
    return CheckoutResponseSchema(
        order_id=charge.reference,
        amount_cents=charge.amount_cents,
        br_code=charge.br_code,
        qr_code_image_url=charge.qr_code_image_url,
        qr_code_base64=charge.qr_code_base64,
    )

PixCharge é schema Pydantic — e é por isso que devolvê-lo vaza

Nada impede um router de anotar -> PixCharge: ele serializa. O problema é o que serializa. Um model_dump(mode="json") de uma cobrança tem 14 campos, e dois deles não são do seu cliente:

['amount_cents', 'br_code', 'currency', 'end_to_end_id', 'expires_at',
 'paid_at', 'provider', 'provider_charge_id', 'provider_status',
 'qr_code_base64', 'qr_code_image_url', 'raw', 'reference', 'status']

raw é o payload cru do provedor — no caminho da OpenPix é o Charge inteiro, customer incluído, com name, email e taxID do pagador (a grafia do fio, que é a que o raw usa). provider_charge_id é a sua chave de escrita no provedor. Um schema de resposta próprio, com os campos que a tela usa, é o que separa a sua API do payload de um terceiro.

Passo 6 — o webhook: verificação na borda, contrato dentro

from fastapi import APIRouter

from src.api.dependencies import CheckoutServiceDep, PixProviderDep, WebhookDeliveryDep

router: APIRouter = APIRouter(prefix="/webhooks", include_in_schema=False)


@router.post("/pix")
async def receive_pix(
    delivery: WebhookDeliveryDep,
    provider: PixProviderDep,
    service: CheckoutServiceDep,
) -> dict[str, str | None]:
    """Turn a verified delivery into a settled order.

    Args:
        delivery (WebhookDeliveryDep): The verified delivery.
        provider (PixProviderDep): The provider that parses it.
        service (CheckoutServiceDep): The service that acts on it.

    Returns:
        dict[str, str | None]: The order settled by this delivery, if any.
    """
    event = provider.parse_webhook(delivery)
    return {"settled": await service.settle(event)}

O router não importa OpenPixWebhookEvent, e não sabe que existe RSA no caminho: ele recebe WebhookDeliveryDep, entrega ao parse_webhook do provider e age sobre o PixPaymentEvent que sai. A verificação de assinatura — a parte que nenhum contrato unifica — ficou inteira dentro do Annotated do composition root.

E ele não libera nada sozinho: service.settle relê a cobrança e marca o pedido sob claim_once (passo 4). Assinatura válida prova que a entrega veio do provedor; quem autoriza mover dinheiro é a releitura. Por isso a mesma entrega, reenviada, responde {"settled": null} na segunda vez.

include_in_schema=False não é cosmético

Webhook não é endpoint da sua API pública: quem se autentica ali é uma assinatura, não o token do seu usuário. Com o router fora do schema, o app.openapi() deste serviço lista uma rota só:

['/api/checkout/{order_id}']

Passo 7 — nos testes, o fake entra pela dependência

O adapter in-memory da última seção desta página não serve só para script: ele entra no lugar do provedor por dependency_overrides, e a suíte inteira roda sem rede.

import asyncio
from pathlib import Path
from typing import Any

from fastapi.testclient import TestClient

from tempest_fastapi_sdk import AsyncDatabaseManager, session_dependency_for
from tempest_fastapi_sdk.integrations.payment import PixProvider

from src.api.app import create_app
from src.api.dependencies import get_pix_provider, get_session, verified_delivery
from tests.fakes import FakePixProvider


def test_checkout_and_webhook(tmp_path: Path) -> None:
    """Charge, settle, and replay the delivery, on the fake and a real SQLite."""
    db = AsyncDatabaseManager(f"sqlite+aiosqlite:///{tmp_path / 'orders.db'}")
    asyncio.run(db.create_tables())

    app = create_app()
    provider: PixProvider = FakePixProvider()
    app.dependency_overrides[get_pix_provider] = lambda: provider
    app.dependency_overrides[get_session] = session_dependency_for(lambda: db)

    def fake_delivery() -> Any:
        """Stand in for the verified delivery.

        Returns:
            Any: What this provider's parse_webhook reads.
        """
        return {"charge_id": "fake-1"}

    app.dependency_overrides[verified_delivery] = fake_delivery

    with TestClient(app) as client:
        created = client.post("/api/checkout/pedido-1042", json={"amount_cents": 1990})
        assert created.status_code == 201
        assert created.json()["br_code"] == "000201fake-1"

        first = client.post("/webhooks/pix")
        assert first.json() == {"settled": "pedido-1042"}

        replay = client.post("/webhooks/pix")
        assert replay.json() == {"settled": None}

As três respostas, rodando:

POST /api/checkout/pedido-1042 -> 201 {'order_id': 'pedido-1042', 'amount_cents': 1990, 'br_code': '000201fake-1', 'qr_code_image_url': None, 'qr_code_base64': None}
POST /webhooks/pix -> 200 {'settled': 'pedido-1042'}
POST /webhooks/pix -> 200 {'settled': None}

A terceira linha é a reentrega: o fake responde PAID de novo na releitura, mas paid_at já está preenchido e o claim_once devolve False.

São três overrides, e eles são diferentes de propósito. O provider troca o provedor inteiro pelo fake. A entrega verificada troca só o verificador — porque assinar é a parte que o fake não tem como imitar, e fingir a verificação em teste é melhor do que desligá-la em produção. A sessão aponta para um SQLite num diretório temporário: o claim_once é um UPDATE de verdade, e testá-lo contra um dict não provaria nada.

O que o type-checker cobra do seu fake

A linha provider: PixProvider = FakePixProvider() é o que faz o mypy --strict conferir o fake contra o contrato. Trocando o parâmetro de create_pix_charge para str, a reprovação é imediata e aponta o campo:

tests/fakes.py:43: error: "str" has no attribute "reference"  [attr-defined]
tests/fakes.py:44: error: "str" has no attribute "amount_cents"  [attr-defined]
tests/test_checkout.py:21: error: Incompatible types in assignment (expression has type "FakePixProvider", variable has type "PixProvider")  [assignment]
tests/test_checkout.py:21: note: Following member(s) of "FakePixProvider" have conflicts:
tests/test_checkout.py:21: note:     Expected:
tests/test_checkout.py:21: note:         def create_pix_charge(self, request: PixChargeRequest) -> Coroutine[Any, Any, PixCharge]
tests/test_checkout.py:21: note:     Got:
tests/test_checkout.py:21: note:         def create_pix_charge(self, request: str) -> Coroutine[Any, Any, PixCharge]

Sem a anotação, dependency_overrides aceita qualquer callable e o defeito só aparece na primeira cobrança.

O que a troca de provedor custa neste serviço

Uma varredura pelo nome do provedor no serviço acima acha dois arquivos:

src/core/settings.py
src/api/dependencies/payments.py

O primeiro só porque as credenciais são mesmo do provedor. O segundo é o composition root — e é a linha OpenPixPixProvider(OpenPixClient(http)) que muda quando o adapter for outro. Nem o service, nem os routers, nem os schemas aparecem nessa lista: é assim que se mede se a costura está no lugar.

Ligando o segundo provedor sem desligar o primeiro

Trocar o provedor num deploy só é apostar em tudo que o sandbox não mostrou. O caminho seguro é um rollout: uma fatia dos pedidos novos vai para o provedor novo, toda cobrança já aberta continua sendo lida e cancelada onde nasceu, e a fatia cresce (ou volta a zero) sem deploy. É o que o PixProviderRouter faz:

import asyncio

from tempest_fastapi_sdk.integrations.payment import (
    PixChargeRequest,
    PixProviderRouter,
)
from tempest_fastapi_sdk.testing.fakes import FakePixProvider


async def main() -> None:
    """Send 10% of new charges to the candidate, read each where it was born."""
    router: PixProviderRouter = PixProviderRouter(
        [
            FakePixProvider(provider_name="openpix"),
            FakePixProvider(provider_name="mercado_pago"),
        ],
        primary="openpix",
        candidate="mercado_pago",
        candidate_percent=10,
    )
    charge = await router.create_pix_charge(
        PixChargeRequest(amount_cents=1990, reference="pedido-1000"),
    )
    again = await router.get_pix_charge(charge.provider, charge.provider_charge_id)
    print(charge.provider, again.provider_charge_id == charge.provider_charge_id)


asyncio.run(main())

O script imprime mercado_pago True: bucket("pedido-1000") é 3, abaixo dos 10 da fatia, então a cobrança nasceu no candidato e foi relida lá.

Três decisões estão nesse objeto:

  1. A fatia é função da reference, não sorteio. bucket(reference) é um SHA-256 reduzido a 0..99, igual em todo processo. O mesmo pedido, reaberto depois de um timeout, cai no mesmo provedor, e subir a porcentagem só acrescenta pedidos à fatia do candidato.
  2. Leitura e cancelamento vão pelo provedor guardado. get_pix_charge e cancel_pix_charge recebem charge.provider além do id. Por isso o pedido guarda as duas colunas, provider e provider_charge_id. O modelo do passo 3 guarda só o id, e com dois provedores ele precisa da outra.
  3. candidate_percent=0 é o kill switch. Nada novo vai para o candidato, e o que já está lá continua sendo servido, porque o provedor segue registrado.

O router não é um PixProvider: o get_pix_charge do contrato recebe só o id, e um id sozinho não diz a quem perguntar. Para o confirm_pix_payment, passe router.provider_for(order.provider).

Cada provedor tem a sua rota de webhook

A verificação é por provedor, e a entrega do Mercado Pago não traz o estado da order: só o id, assinado. A dependency make_mercado_pago_webhook_delivery_dependency verifica a assinatura e relê a order antes de entregar ao parse_webhook, e o parse_webhook do adapter recusa a notificação crua. Monte /webhooks/openpix e /webhooks/mercado-pago lado a lado, cada um com a sua dependency, e o mesmo service.settle atrás dos dois. Detalhes em Mercado Pago ».

Estados

Você decide sobre PaymentStatus, não sobre a string do provedor:

canônico significa
PENDING criada, esperando o pagador
AUTHORIZED cartão aprovado pelo emissor e retido, ainda não capturado
PAID liquidada
EXPIRED a janela fechou sem pagamento
CANCELLED retirada por você ou pelo provedor
REFUNDED paga e devolvida
CHARGED_BACK revertida pela instituição do pagador
IN_ANALYSIS retida para revisão
FAILED recusada
UNKNOWN um estado que esta versão do SDK não classifica

A string original não se perde: ela fica em provider_status, que é o que você põe no log e mostra no suporte.

from tempest_fastapi_sdk.integrations.payment import PaymentStatus, PixCharge


def liberar_pedido(charge: PixCharge) -> bool:
    """Decide whether the order can be released.

    Args:
        charge (PixCharge): The charge as get_pix_charge re-read it, never
            the copy a webhook delivered.

    Returns:
        bool: Whether the money is in.
    """
    return charge.status is PaymentStatus.PAID

is funciona aqui — e não funcionaria de graça

PixCharge desliga o use_enum_values que o BaseSchema liga. Sem isso, charge.status guardaria a string "paid" e charge.status is PaymentStatus.PAID seria False em toda cobrança, silenciosamente, enquanto == continuaria funcionando. É o tipo de defeito que sobrevive à revisão justamente porque o teste óbvio passa.

O provedor pode inventar um estado — e isso não derruba a leitura

UNKNOWN existe porque um provedor adiciona estado sem avisar. As duas formas de esconder isso são piores: cair em PENDING afirma que a cobrança espera pagamento logo depois de o provedor dizer que não; e recusar a leitura transforma um estado que o SDK não conhece numa requisição falhada, com o estado real em lugar nenhum que você veja.

A string original fica em provider_status, então dá para ramificar nela enquanto o mapeamento canônico não existe:

if charge.status is PaymentStatus.UNKNOWN:
    logger.warning("estado novo da OpenPix: %s", charge.provider_status)

Webhook

A verificação de assinatura continua sendo de cada provedor — RSA-1024 no OpenPix, HMAC no Stripe. O que o contrato unifica é o que sai dela:

from tempest_fastapi_sdk.integrations.payment import PixEventType, PixProvider
from tempest_fastapi_sdk.integrations.payment.openpix import OpenPixWebhookEvent


def tratar(provider: PixProvider, delivery: OpenPixWebhookEvent) -> str | None:
    """Find which order a verified delivery is about.

    Args:
        provider (PixProvider): The provider that verified the delivery.
        delivery (OpenPixWebhookEvent): The verified event.

    Returns:
        str | None: The reference of the order to re-check, if any.
    """
    event = provider.parse_webhook(delivery)
    if event.type is PixEventType.CHARGE_PAID and event.charge is not None:
        return event.charge.reference
    return None

O que sai daí é qual pedido conferir, não um pedido pago. Liberar é o settle do passo 4: relê a cobrança com confirm_pix_payment e marca sob claim_once.

Um evento que o SDK não classifica vira PixEventType.UNKNOWN com o nome original preservado em provider_event_name. Ele fica visível, não engolido.

Os tipos canônicos são seis:

PixEventType disparado quando
CHARGE_CREATED a cobrança foi aberta
CHARGE_PAID o dinheiro entrou
CHARGE_EXPIRED a janela fechou sem pagamento
CHARGE_CANCELLED a cobrança foi retirada
CHARGE_REFUNDED o valor foi devolvido
UNKNOWN o provedor mandou algo que o SDK não mapeia

Por que parse_webhook recebe Any

Cada provedor entrega um tipo diferente: o OpenPix entrega OpenPixWebhookEvent, já verificado; outro provedor entregaria o dict do corpo, ou um objeto próprio. Tipar o parâmetro como o tipo de um provedor amarraria o contrato a ele — que é exatamente o acoplamento que esta página evita. O adapter conhece o seu tipo; o contrato conhece só o que sai, que é PixPaymentEvent.

Verificar a assinatura vem antes e continua sendo do provedor: RSA-1024 no OpenPix, HMAC no Stripe. O passo a passo do lado OpenPix, incluindo como registrar a URL, está em OpenPix ».

O que o provedor diz além do contrato

Fica em raw:

from tempest_fastapi_sdk.integrations.payment import PixCharge


def link_de_pagamento(charge: PixCharge) -> object | None:
    """Read a provider-specific field the contract does not model.

    Args:
        charge (PixCharge): The charge.

    Returns:
        object | None: OpenPix's payment link, when present.
    """
    return charge.raw.get("paymentLinkUrl")

Por que raw precisa existir

O BaseSchema do SDK é extra="ignore". Sem esse campo, tudo que o provedor manda além do contrato sumiria na validação — sem erro, sem aviso. raw é o que garante que passar pelo contrato não perde informação.

O raw usa a grafia do fio nos dois caminhos, e é por isso que a busca acima é raw["paymentLinkUrl"] e não raw["payment_link_url"]. O nome que o provedor documenta é o único que você consegue prever a partir da documentação dele.

Mudou na v0.270.0

Até a v0.269.0, o caminho de API dumpava sem by_alias, então campo declarado saía em snake_case enquanto campo não-declarado (que extra="allow" preserva) saía em camelCase — o raw era uma mistura, e raw["paymentLinkUrl"] devolvia None em toda cobrança lida pela API. Serviço que já lia raw com snake_case precisa trocar para o nome do fio.

A admonition anterior também explicava isso errado. Ela dizia que no caminho de API o raw era o payload "depois de o schema validar, então campos não declarados já foram descartados". Isso deixou de valer quando os modelos de resposta viraram extra="allow" (v0.259/v0.260): paidAt sobrevive à validação e está no raw dos dois caminhos. O que continua diferente é só que o adapter da OpenPix lê paid_at na entrega de webhook e não no caminho de API.

Escrevendo um adapter

Um adapter é uma classe com provider_name e os quatro métodos. Nada a herdar. O exemplo abaixo não fala com provedor nenhum — guarda cobranças num dict — e por isso serve para testar o seu service sem rede, que é o primeiro adapter que vale escrever:

import asyncio
from typing import Any

from tempest_fastapi_sdk.integrations.payment import (
    PaymentStatus,
    PixCharge,
    PixChargeRequest,
    PixEventType,
    PixPaymentEvent,
    PixProvider,
)


class FakePixProvider:
    """A provider that keeps charges in a dict instead of calling anyone.

    Attributes:
        provider_name (str): The identifier copied into
            ``PixCharge.provider``.
    """

    provider_name: str = "fake"

    def __init__(self) -> None:
        """Start with no charges."""
        self._charges: dict[str, PixCharge] = {}
        self._next_id: int = 1

    async def create_pix_charge(self, request: PixChargeRequest) -> PixCharge:
        """Create a charge in memory.

        Args:
            request (PixChargeRequest): What the service asked to charge.

        Returns:
            PixCharge: The charge, in canonical shape.
        """
        charge_id = f"fake-{self._next_id}"
        self._next_id += 1
        charge = PixCharge(
            provider=self.provider_name,
            provider_charge_id=charge_id,
            reference=request.reference,
            amount_cents=request.amount_cents,
            status=PaymentStatus.PENDING,
            provider_status="created",
            br_code=f"000201{charge_id}",
        )
        self._charges[charge_id] = charge
        return charge

    async def get_pix_charge(self, charge_id: str) -> PixCharge:
        """Read a charge back.

        Args:
            charge_id (str): The provider-side id.

        Returns:
            PixCharge: The stored charge.

        Raises:
            KeyError: When no charge carries that id.
        """
        return self._charges[charge_id]

    async def cancel_pix_charge(self, charge_id: str) -> PixCharge:
        """Withdraw an unpaid charge.

        Args:
            charge_id (str): The provider-side id.

        Returns:
            PixCharge: The charge in its cancelled shape.
        """
        charge = self._charges[charge_id]
        cancelled = charge.model_copy(
            update={
                "status": PaymentStatus.CANCELLED,
                "provider_status": "cancelled",
            },
        )
        self._charges[charge_id] = cancelled
        return cancelled

    def parse_webhook(self, event: Any) -> PixPaymentEvent:
        """Turn a delivery into a canonical event.

        Args:
            event (Any): Whatever this provider delivers.

        Returns:
            PixPaymentEvent: The canonical event.
        """
        charge = self._charges[str(event["charge_id"])]
        paid = charge.model_copy(
            update={"status": PaymentStatus.PAID, "provider_status": "paid"},
        )
        self._charges[paid.provider_charge_id] = paid
        return PixPaymentEvent(
            provider=self.provider_name,
            type=PixEventType.CHARGE_PAID,
            provider_event_name="fake.paid",
            charge=paid,
            raw=dict(event),
        )


async def main() -> None:
    """Exercise the whole contract against the fake."""
    provider: PixProvider = FakePixProvider()

    charge = await provider.create_pix_charge(
        PixChargeRequest(amount_cents=1990, reference="pedido-1042"),
    )
    print(charge.provider_charge_id, charge.status.value)

    event = provider.parse_webhook({"charge_id": charge.provider_charge_id})
    print(event.type.value, event.charge is not None)


if __name__ == "__main__":
    asyncio.run(main())

Rodando, isso imprime:

fake-1 pending
charge_paid True

A linha que faz o trabalho de verificação é provider: PixProvider = FakePixProvider(). Ela não muda nada em tempo de execução — muda o que o type-checker exige de você. Se um método sair com a assinatura errada, mypy --strict reprova ali, e não no dia da primeira cobrança.

Três coisas que um adapter de provedor real faz a mais

  1. Converte a unidade. O contrato é centavo inteiro; o provedor pode querer reais decimais. A conversão é do adapter, e é por isso que ela fica num lugar só.
  2. Mapeia o estado. A string do provedor vira PaymentStatus, e a original é copiada em provider_status — sem descartar nada.
  3. Preenche raw. Tudo que o provedor diz além do contrato vai para lá, para nenhuma informação morrer na tradução.

O OpenPixPixProvider é a referência de como as três ficam juntas: integrations/payment/adapters/openpix.py.

Recap

  • Seu service depende de PixProvider e recebe PixCharge.
  • O contrato tem quatro métodos: criar, ler, cancelar e interpretar webhook. Guarde provider_charge_id — é o argumento dos dois do meio.
  • Dinheiro atravessa o contrato como int de centavos.
  • O estado que você usa é PaymentStatus; o do provedor fica ao lado, em provider_status.
  • Assinatura de webhook continua por provedor; o evento que sai dela é canônico, e o que o SDK não mapeia chega como UNKNOWN com o nome original.
  • Nada se perde: o que o provedor diz a mais está em raw.
  • Adapter é classe com provider_name e os quatro métodos, sem herança. Hoje o SDK ships dois (OpenPix e Mercado Pago); o fake in-memory acima é o que você escreve primeiro, para testar sem rede.
  • Para migrar de provedor, PixProviderRouter manda uma fatia estável dos pedidos novos ao candidato e lê cada cobrança no provedor que a criou.
  • Na arquitetura: o provider é montado em api/dependencies e sai de lá como PixProvider. Um HTTPClient por processo, no lifespan — e app.state é Any, então reanote o tipo na leitura.
  • O router devolve schema seu, não PixCharge: a cobrança canônica carrega raw (o payload do provedor, customer incluído) e provider_charge_id.
  • O webhook avisa, a API confirma: o settle relê a cobrança pelo provider_charge_id guardado com confirm_pix_payment (deste pedido, PAID, valor igual) e libera sob claim_once, então a reentrega não paga duas vezes.
  • No teste, dependency_overrides troca o provider pelo fake, a entrega verificada por um stub e a sessão por um SQLite temporário — três overrides, e a suíte roda sem rede.