# Recipe: Django

For Django 4.2 or later (Python 3.10+), with Django templates. The API client uses only the standard library. This file has the code; [the playbook](https://checkouts.md/playbook.md) has the reasons, every error code and what to tell the person. Follow both.

| | |
|---|---|
| Server code runs in | Views in a small `payments` app |
| Database | The project's database through the ORM |
| Webhook path | `/webhooks/railbed` (exempt from CSRF) |
| Connect | `npx checkouts.md login` (Test into `.env`, Live into `.env.production`) |

## What this adds

| File | What it does |
|---|---|
| `payments/models.py` | `Order` and `RailbedEvent` (tables `orders` and `railbed_events`) |
| `payments/railbed.py` | Products, the API client and signature verification |
| `payments/views.py` | The checkout, the success page and the webhook |
| `payments/urls.py` | The three routes |
| `payments/templates/payments/success.html`, the Pay button | The templates |
| `settings.py`, the root `urls.py` | The settings and the app's routes |
| `CHECKOUTS.md` | How this project takes payments |

If the project already has an `Order` model or an `orders` table, call these `CheckoutOrder` and `checkout_orders` (or add the fields to the existing model).

## Environment

| Name | Development (`.env`, gitignored) | Production (the host's environment) |
|---|---|---|
| `RAILBED_SECRET_KEY` | `rb_test_…` | `rb_live_…` |
| `RAILBED_WEBHOOK_SECRET` | `whsec_…` of the Test endpoint | `whsec_…` of the Live endpoint |
| `APP_URL` | `http://localhost:8000` | `https://yourapp.com` |

Django doesn't read `.env` by itself. Use what the project already uses (`django-environ`, `python-dotenv`), or export the file into the shell without printing it: `set -a; . ./.env; set +a`.

## 1. Connect

```bash
npx checkouts.md login
```

The Test key goes to `.env` and the Live key to `.env.production`, both gitignored. Relay the link and code to the person and wait for the command to finish.

## 2. Settings and database

`settings.py`:

```python
import os

INSTALLED_APPS = [
    # …the project's apps…
    "payments",
]

RAILBED_SECRET_KEY = os.environ.get("RAILBED_SECRET_KEY", "")
RAILBED_WEBHOOK_SECRET = os.environ.get("RAILBED_WEBHOOK_SECRET", "")
RAILBED_API_BASE = os.environ.get("RAILBED_API_BASE", "https://pay.railbed.io/v1")
APP_URL = os.environ.get("APP_URL", "http://localhost:8000").rstrip("/")

# Django drops INFO lines from the project's own loggers unless LOGGING says otherwise. This shows the payments app's,
# including the webhook's result that step 5 checks; merge it into the project's LOGGING if there is one.
LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "handlers": {"console": {"class": "logging.StreamHandler"}},
    "loggers": {"payments": {"handlers": ["console"], "level": "INFO"}},
}
```

`payments/models.py`:

```python
from django.conf import settings
from django.db import models


class Order(models.Model):
    id = models.CharField(primary_key=True, max_length=64)  # order_…: sent as reference and in the Idempotency-Key
    product_id = models.CharField(max_length=64)
    amount = models.CharField(max_length=20)  # "49.00": the exact decimal string sent to Railbed
    currency = models.CharField(max_length=3)
    livemode = models.BooleanField()
    status = models.CharField(max_length=20, default="awaiting_payment")
    payment_id = models.CharField(max_length=64, unique=True, null=True, blank=True)  # pay_…
    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True, on_delete=models.SET_NULL)
    created_at = models.DateTimeField(auto_now_add=True)
    paid_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        db_table = "orders"


class RailbedEvent(models.Model):
    id = models.CharField(primary_key=True, max_length=64)  # evt_…: each webhook event is handled once
    type = models.CharField(max_length=40)
    payment_id = models.CharField(max_length=64, null=True, blank=True)
    received_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = "railbed_events"
```

```bash
python manage.py makemigrations payments
python manage.py migrate
```

## 3. The code

`payments/railbed.py`:

```python
import hashlib
import hmac
import json
import re
import time
import urllib.error
import urllib.request

from django.conf import settings

# Prices live here, on the server, as decimal strings (R3). The browser only ever sends an id.
PRODUCTS = {
    "pro": {"id": "pro", "name": "Pro plan", "amount": "49.00", "currency": "USD"},
}


class RailbedError(Exception):
    def __init__(self, status: int, code: str, message: str):
        super().__init__(message)
        self.status = status
        self.code = code


def railbed(path: str, body: dict | None = None, idempotency_key: str | None = None) -> dict:
    """Calls the Railbed API with the secret key. Server code only."""
    if not settings.RAILBED_SECRET_KEY:
        raise RuntimeError("RAILBED_SECRET_KEY is not set")
    headers = {
        "Authorization": f"Bearer {settings.RAILBED_SECRET_KEY}",
        "Accept": "application/json",
        "User-Agent": "checkouts.md-django",
    }
    data = None
    if body is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(body).encode()
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key
    request = urllib.request.Request(
        settings.RAILBED_API_BASE.rstrip("/") + path,
        data=data,
        headers=headers,
        method="GET" if body is None else "POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return json.loads(response.read())
    except urllib.error.HTTPError as error:
        try:
            detail = json.loads(error.read()).get("error", {})
        except ValueError:
            detail = {}
        raise RailbedError(error.code, detail.get("code", "unknown"), detail.get("message", f"Railbed answered {error.code}")) from None


def is_live_key() -> bool:
    return settings.RAILBED_SECRET_KEY.startswith("rb_live_")


def verify_railbed_signature(raw_body: bytes, header: str, secret: str, tolerance_seconds: int = 300) -> bool:
    """Railbed-Signature: t=<seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">, one or more v1 (two during a secret roll)."""
    match = re.fullmatch(r"t=(\d{1,12})((?:,v1=[0-9a-f]{64})+)", header or "")
    if not match:
        return False
    timestamp, signatures = match.groups()
    if abs(time.time() - int(timestamp)) > tolerance_seconds:
        return False
    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, part[3:]) for part in signatures[1:].split(","))
```

`payments/views.py`:

```python
import json
import logging
import time
import uuid
from datetime import datetime, timezone
from urllib.parse import quote

from django.conf import settings
from django.db import IntegrityError, transaction
from django.http import HttpResponse, HttpResponseBadRequest, HttpResponseRedirect, JsonResponse
from django.shortcuts import render
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from .models import Order, RailbedEvent
from .railbed import PRODUCTS, RailbedError, is_live_key, railbed, verify_railbed_signature

logger = logging.getLogger(__name__)

MESSAGES = {
    "paid": "Payment received. Your order is confirmed.",
    "awaiting_payment": "Confirming your payment…",
    "held": "Your payment is being reviewed. We’ll email you once it’s done.",
    "failed": "This payment didn’t complete.",
    "expired": "This payment didn’t complete.",
}


@require_POST
def checkout(request):
    wants_json = request.content_type == "application/json"
    try:
        data = json.loads(request.body or b"{}") if wants_json else request.POST
    except ValueError:
        data = {}
    product = PRODUCTS.get(str(data.get("product", ""))) if hasattr(data, "get") else None
    if product is None:
        return JsonResponse({"error": "Unknown product."}, status=400)

    user = getattr(request, "user", None)
    order = Order.objects.create(
        id=f"order_{uuid.uuid4()}",
        product_id=product["id"],
        amount=product["amount"],
        currency=product["currency"],
        livemode=is_live_key(),
        user=user if user is not None and user.is_authenticated else None,
    )
    try:
        session = railbed(
            "/checkout_sessions",
            body={
                "amount": product["amount"],
                "currency": product["currency"],
                "description": product["name"],
                "reference": order.id,
                "metadata": {"order_id": order.id, "product_id": product["id"]},
                "success_url": settings.APP_URL + "/checkout/success?order={REFERENCE}",
                "cancel_url": settings.APP_URL + "/",
            },
            idempotency_key=order.id,
        )
    except (RailbedError, OSError) as error:  # OSError: no connection, or a timeout
        logger.error("Railbed: checkout session not created %s %s", order.id, getattr(error, "code", error))
        return JsonResponse({"error": "We couldn't start the payment. Please try again."}, status=502)

    Order.objects.filter(id=order.id).update(payment_id=session["id"])
    return JsonResponse({"url": session["url"]}) if wants_json else HttpResponseRedirect(session["url"])


@require_GET
def success(request):
    """Shows the order's status from the database. It never marks anything paid: only the webhook does (R2)."""
    order = Order.objects.filter(id=request.GET.get("order", "")).first()
    response = render(
        request,
        "payments/success.html",
        {
            "message": MESSAGES.get(order.status, "Checking your order…") if order else "We couldn’t find that order.",
            "waiting": order is not None and order.status == "awaiting_payment",
            "retry": order is not None and order.status in ("failed", "expired"),
        },
    )
    response["Cache-Control"] = "no-store"
    return response


def fulfil(order: Order) -> None:
    """Grants what the order bought. Runs once per order, inside the transaction that marks it paid.

    This is the one project-specific piece: replace the body, for example
        Profile.objects.filter(user_id=order.user_id).update(plan=order.product_id)
    Queue slow work (emails, shipping labels) with transaction.on_commit(...) so it runs only after the commit.
    """
    logger.info("fulfil %s (%s)", order.id, order.product_id)


def apply_payment(order: Order, payment: dict) -> str:
    if payment["status"] == "paid":
        paid_at = datetime.fromtimestamp(payment.get("paid_at") or time.time(), tz=timezone.utc if settings.USE_TZ else None)
        changed = Order.objects.filter(id=order.id).exclude(status="paid").update(status="paid", paid_at=paid_at)
        if changed == 1:
            fulfil(Order.objects.get(id=order.id))
            return "fulfilled"
    elif payment["status"] in ("held", "failed", "expired"):
        Order.objects.filter(id=order.id).exclude(status="paid").update(status=payment["status"])
    return "recorded"


@csrf_exempt  # Railbed's request is signed, not a form
@require_POST
def railbed_webhook(request):
    raw = request.body  # the exact bytes Railbed signed: verify before parsing
    secret = settings.RAILBED_WEBHOOK_SECRET
    if not secret or not verify_railbed_signature(raw, request.headers.get("Railbed-Signature", ""), secret):
        return HttpResponseBadRequest("Invalid signature")

    event = json.loads(raw)
    ref = (event.get("data") or {}).get("payment") or {}
    if not str(event.get("type", "")).startswith("payment.") or not ref.get("id"):
        return HttpResponse(status=200)
    if event.get("livemode") is not is_live_key():
        logger.error("Railbed webhook: mode mismatch. RAILBED_SECRET_KEY and RAILBED_WEBHOOK_SECRET must be the same mode.")
        return HttpResponse("Mode mismatch", status=500)

    try:
        payment = railbed(f"/payments/{quote(ref['id'], safe='')}")
    except RailbedError as error:
        # "Send test event" samples carry a made-up payment the API doesn't know: acknowledge them.
        sample = not event["livemode"] and (ref.get("metadata") or {}).get("sample") == "true"
        if error.status == 404 and sample:
            return HttpResponse(status=200)
        logger.error("Railbed webhook: payment not read %s %s", event.get("id"), error.code)
        return HttpResponse("Try again later", status=500)

    order = Order.objects.filter(payment_id=payment["id"]).first()
    if order is None:
        return HttpResponse(status=200)  # not a payment this app created
    if (
        payment["livemode"] != order.livemode
        or payment["reference"] != order.id
        or payment["amount"] != order.amount
        or payment["currency"] != order.currency
    ):
        logger.error("Railbed webhook: payment %s does not match order %s", payment["id"], order.id)
        return HttpResponse(status=200)

    # Exactly once (playbook 6.4, step 8): the event, the paid update and the grant commit together. Only the event
    # insert is allowed to fail as a duplicate (inside its own savepoint); any other error rolls back and answers 500.
    with transaction.atomic():
        try:
            with transaction.atomic():
                RailbedEvent.objects.create(id=event["id"], type=event["type"], payment_id=payment["id"])
        except IntegrityError:
            result = "duplicate"  # this event id was already handled
        else:
            result = apply_payment(order, payment)
    logger.info("Railbed webhook %s %s: %s", event["type"], event["id"], result)
    return HttpResponse(status=200)
```

`payments/urls.py`:

```python
from django.urls import path

from . import views

urlpatterns = [
    path("checkout/", views.checkout, name="checkout"),
    path("checkout/success", views.success, name="checkout_success"),
    path("webhooks/railbed", views.railbed_webhook, name="railbed_webhook"),
]
```

The root `urls.py`: add `path("", include("payments.urls"))` to `urlpatterns` (with `include` from `django.urls`). Register the webhook address exactly as `/webhooks/railbed`: with `APPEND_SLASH`, a POST to an address that only matches with a trailing slash is redirected (an error page with `DEBUG` on), and Railbed counts a redirect as a failed delivery.

`payments/templates/payments/success.html`:

```html
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  {% if waiting %}<meta http-equiv="refresh" content="3">{% endif %}
  <title>Your order</title>
</head>
<body>
  <main>
    <h1>{{ message }}</h1>
    {% if waiting %}<p>This usually takes a few seconds.</p>{% endif %}
    {% if retry %}<p><a href="/">Try again</a></p>{% endif %}
  </main>
</body>
</html>
```

The Pay button, in any template:

```html
<form method="post" action="{% url 'checkout' %}">
  {% csrf_token %}
  <input type="hidden" name="product" value="pro">
  <button type="submit">Pay $49.00</button>
</form>
```

## 4. The webhook endpoint

```bash
set -a; . ./.env; set +a
python manage.py runserver                       # http://localhost:8000
cloudflared tunnel --url http://localhost:8000   # in a second terminal; prints https://….trycloudflare.com
npx checkouts.md webhook https://<tunnel>.trycloudflare.com/webhooks/railbed
```

Add the tunnel's host to `ALLOWED_HOSTS` while testing (for example `".trycloudflare.com"`). The CLI adds `RAILBED_WEBHOOK_SECRET` to `.env`; load the file again and restart `runserver`.

## 5. Test it

1. Create an order through the app's own route: open the page with the Pay button and click it; the address bar shows `https://pay.railbed.io/p/pay_…`.
2. Pay it: choose **Simulate successful payment** on the test provider page, or run the playbook's `simulate.mjs` with the `pay_…` id ([7.1](https://checkouts.md/playbook.md#71-the-end-to-end-run)).
3. Check: the log shows `Railbed webhook payment.paid evt_…: fulfilled`; `python manage.py shell -c "from payments.models import Order; print(Order.objects.get(payment_id='pay_…').status)"` prints `paid`; `/checkout/success?order=order_…` says it's confirmed.
4. Run the other scenarios in the playbook's [7.2](https://checkouts.md/playbook.md#72-the-scenarios), including `replay.mjs`.
5. Record the ids under Evidence in `CHECKOUTS.md`, and delete the tunnel's endpoint.

## 6. Go live

1. Put `RAILBED_SECRET_KEY` (from `.env.production`) and `APP_URL=https://…` into the production environment without printing the key, and run `python manage.py migrate` there.
2. Deploy. The production domain is in `ALLOWED_HOSTS`.
3. Add the Live endpoint: `npx checkouts.md webhook https://yourapp.com/webhooks/railbed --live` (plus the same `--live-to` as when connecting, if you gave one). It writes the Live `RAILBED_WEBHOOK_SECRET` into the Live key's file; move it to production the same way and restart.
4. Check: `npx checkouts.md status` (with that `--live-to`, if any); `curl -s -w ' %{http_code}\n' -X POST https://yourapp.com/webhooks/railbed -d '{}'` prints `Invalid signature 400` (a `403` means the CSRF exemption is missing; a `400` without `Invalid signature`, a host missing from `ALLOWED_HOSTS`).
5. Set `mode: live` in `CHECKOUTS.md`, add the Live endpoint's id under Wiring and the checks under Evidence. The first real payment is the final evidence.

## 7. CHECKOUTS.md

Fill in [the template](https://checkouts.md/templates/CHECKOUTS.md) with `webhook: /webhooks/railbed`. The Wiring section for this recipe:

```markdown
## Wiring

- `payments/railbed.py`: the products and their prices, the API client and signature verification.
- `payments/views.py`: `checkout` saves the order and creates the checkout session (Idempotency-Key = order id); `success` shows status only, never marks paid; `railbed_webhook` verifies, re-reads the payment and applies it once in a transaction (`fulfil()` is the grant).
- `payments/models.py`: `Order` (`orders`) and `RailbedEvent` (`railbed_events`).
- `payments/urls.py`: `/checkout/`, `/checkout/success`, `/webhooks/railbed` (exempt from CSRF).
- Webhook endpoints: Test `we_…` (tunnel, removed after testing), Live `we_…`, at `/webhooks/railbed`.
- Secrets: `RAILBED_SECRET_KEY`, `RAILBED_WEBHOOK_SECRET`. Test in `.env` (gitignored); Live in the production environment.
```

## Notes

- **Flask:** the same `railbed.py` without Django settings (read `os.environ`), the raw body from `request.get_data()`, and the header from `request.headers.get("Railbed-Signature", "")`.
- **FastAPI:** `raw = await request.body()` in an `async def` route taking `request: Request`; run the blocking `urllib` call with `await asyncio.to_thread(railbed, path)`, or use `httpx.AsyncClient`.
- **Django REST Framework:** a DRF view parses the body for you; keep the webhook a plain Django view, as above, so `request.body` is untouched.