An async FastAPI payment flow built for exactness and retries

Step 1 · configuration and money

Fail fast on missing secrets and never calculate with float

Environment files are suitable for local development; use your platform’s encrypted secret store in hosted environments. Keep integer paise in storage and convert through Decimal only where the API accepts rupees.

# requirements
# fastapi, httpx, pydantic-settings, uvicorn

from decimal import Decimal

import httpx
from pydantic import AnyHttpUrl, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",
    )

    vg_base_url: AnyHttpUrl
    vg_api_key: SecretStr
    vg_webhook_secret: SecretStr
    public_store_url: AnyHttpUrl


settings = Settings()
gateway_timeout = httpx.Timeout(10.0, connect=3.0)


def paise_to_rupees(paise: int) -> str:
    if paise <= 0:
        raise ValueError("Amount must be positive")
    value = Decimal(paise) / Decimal(100)
    return format(value.quantize(Decimal("0.01")), "f")


def rupees_to_paise(value: str | Decimal) -> int:
    decimal_value = Decimal(value).quantize(Decimal("0.01"))
    paise = decimal_value * Decimal(100)
    if paise != paise.to_integral_value():
        raise ValueError("Amount has more than two decimal places")
    return int(paise)

Step 3 · checkout route

Price and persist before creating a payment

The repository, customer, cart, and catalogue services are application boundaries: they must enforce tenant ownership, database uniqueness, and trusted pricing. The payment logic shown here does not accept a browser-supplied total.

from fastapi import Depends, FastAPI, HTTPException, Request, status
from pydantic import BaseModel, Field

app = FastAPI()


class CheckoutRequest(BaseModel):
    cart_id: str = Field(min_length=1, max_length=64)


@app.post("/api/checkout", status_code=status.HTTP_201_CREATED)
async def checkout(
    payload: CheckoutRequest,
    request: Request,
    customer: Customer = Depends(require_customer),
):
    cart = await carts.load_owned(payload.cart_id, customer.id)
    priced_cart = await pricing.price_from_catalogue(cart)

    # The repository enforces UNIQUE(customer_id, cart_id) and
    # UNIQUE(client_txn_id); a retry loads the same pending order.
    order = await orders.insert_or_load_pending(
        cart_id=cart.id,
        customer_id=customer.id,
        amount_paise=priced_cart.total_paise,
    )

    try:
        gateway = await create_gateway_order(
            client_txn_id=order.client_txn_id,
            amount_paise=order.amount_paise,
            description=f"Order {order.public_number}",
            customer={
                "name": customer.name,
                "email": customer.email,
                "mobile": customer.mobile,
            },
        )
        await orders.attach_gateway_order(
            order.id,
            gateway_order_id=gateway["order_id"],
            payment_url=gateway["payment_url"],
            expires_at=gateway["expires_at"],
        )
    except (httpx.HTTPError, RuntimeError) as exc:
        request.app.state.logger.exception(
            "checkout failed",
            extra={"order_id": str(order.id)},
        )
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="Payment checkout is temporarily unavailable",
        ) from exc

    return {
        "order_id": order.public_number,
        "payment_url": gateway["payment_url"],
        "expires_at": gateway["expires_at"],
    }

Step 5 · reconciliation

Recover uncertain states from a controlled worker

Reconcile ambiguous create calls, webhook outages, and overdue pending orders with bounded retries and backoff. Apply the same immutable-field checks and state machine as the webhook consumer.

async def check_gateway_order(
    *,
    order_id: str | None = None,
    client_txn_id: str | None = None,
) -> dict[str, Any]:
    if not order_id and not client_txn_id:
        raise ValueError("An order identifier is required")

    async with httpx.AsyncClient(timeout=gateway_timeout) as client:
        response = await client.post(
            (
                f"{str(settings.vg_base_url).rstrip('/')}"
                "/api/v1/check_order_status"
            ),
            headers={
                "Content-Type": "application/json",
                "X-API-Key": settings.vg_api_key.get_secret_value(),
            },
            json={
                "order_id": order_id,
                "client_txn_id": client_txn_id,
            },
        )

    payload = response.json()
    data = payload.get("data")
    if response.is_error or payload.get("status") is not True or not data:
        raise RuntimeError(
            str(payload.get("msg") or "Unable to reconcile payment")
        )
    return data

Questions

FastAPI integration FAQ

Should a FastAPI application use float for rupee amounts?

No. Store integer paise and use Decimal only at the API boundary. A binary float can introduce rounding differences that are unacceptable when matching a payment to an order.

Can I use request.json() before webhook verification?

Read await request.body() first and verify the signature over those exact bytes. Parsing and re-encoding the body before verification may change its byte representation.

Should webhook fulfilment run in FastAPI BackgroundTasks?

Not for work that must survive a process restart. Commit the payment and an outbox record in one database transaction, then use a durable worker to perform fulfilment.

How do I prevent duplicate payment processing?

Use database uniqueness for the client transaction ID and verified event key, lock the mapped order during transition, and make paid-to-paid delivery a no-op.

What should the customer-facing status endpoint return?

Return your own order state after authentication and ownership checks. Do not expose the API key, webhook secret, raw provider payload, or an unrestricted gateway status proxy.