Build a payment gateway

In this guide you add a gateway called Acme Pay to BWH Payments. By the end, shoppers can pay through Acme's hosted checkout, a signed webhook marks the payment Paid, and refunds reach Acme Pay. Tests cover all of it without calling the real API.

What you'll learn

• How to split a gateway into a controller and a service client
• How to implement the methods of PaymentGatewayBase
• How to verify webhooks safely
• How to test the gateway against an in-memory Acme Pay

Requirements

• A Frappe 16+ bench with BWH Payments installed
• Your own Frappe app to hold the gateway, for example acme_payments
• Acme's API docs, and a sandbox API key and webhook secret

Add BWH Payments to your app's hooks.py, so it's always installed first:

required_apps = ["bwh_payments"]

How the parts fit

Service clients explains this split.

Step 1: Create the settings DocType

Turn on developer mode, so Frappe writes the DocType into your app:

bench set-config -g developer_mode 1

Create a new DocType in the desk:

• Name: Acme Pay Gateway Settings
• Module: a module in your app
• Is Single: ✅
• Permissions: System Manager only, because it stores secrets

Add these fields:

Step 2: Describe Acme's API

Create acme_payments/services/acme/base.py. It lists the calls the controller needs, and models only the fields it reads:

from abc import ABC, abstractmethod

from bwh_payments.services import GatewayModel


class AcmeCheckout(GatewayModel):
	id: str
	hosted_url: str | None = None
	status: str | None = None


class AcmeRefund(GatewayModel):
	id: str
	status: str | None = None
	amount: int  # minor units, as Acme echoes it


class BaseAcmeClient(ABC):
	"""A rejected call raises frappe.ValidationError."""

	@abstractmethod
	def create_checkout(self, payload: dict) -> AcmeCheckout: ...

	@abstractmethod
	def get_checkout(self, checkout_id: str) -> AcmeCheckout: ...

	@abstractmethod
	def create_refund(self, checkout_id: str, amount: int) -> AcmeRefund: ...

GatewayModel ignores fields you didn't declare, and reads a bare number into a str field. Make a field optional wherever Acme can send null.

Step 3: Write the live client

services/acme/live.py makes the calls. It logs each one by endpoint only, and turns a refusal into a readable error:

import frappe
from frappe import _
from frappe.integrations.utils import create_request_log, make_get_request, make_post_request

from acme_payments.services.acme.base import AcmeCheckout, AcmeRefund, BaseAcmeClient

ACME_API = "https://api.acmepay.example/v1"


class LiveAcmeClient(BaseAcmeClient):
	def __init__(self, api_key: str):
		self.headers = {"Authorization": f"Bearer {api_key}"}

	def create_checkout(self, payload: dict) -> AcmeCheckout:
		return AcmeCheckout.model_validate(self.request("POST", "/checkouts", payload))

	def get_checkout(self, checkout_id: str) -> AcmeCheckout:
		return AcmeCheckout.model_validate(self.request("GET", f"/checkouts/{checkout_id}"))

	def create_refund(self, checkout_id: str, amount: int) -> AcmeRefund:
		return AcmeRefund.model_validate(
			self.request("POST", f"/checkouts/{checkout_id}/refunds", {"amount": amount})
		)

	def request(self, method: str, path: str, payload: dict | None = None) -> dict:
		url = f"{ACME_API}{path}"
		try:
			if method == "GET":
				response = make_get_request(url, headers=self.headers)
			else:
				response = make_post_request(url, json=payload, headers=self.headers)
		except Exception as exception:
			log(url, error=str(exception))
			frappe.throw(_("Acme Pay rejected the request"))
		log(url)
		return response


def log(url: str, error: str | None = None):
	# Payloads carry the shopper's details, so only the endpoint is logged.
	create_request_log(
		{"endpoint": url},
		service_name="Acme Pay",
		is_remote_request=True,
		error=error,
		status="Failed" if error else "Completed",
	)

Step 4: Write the stub

services/acme/stub.py behaves like Acme Pay, in memory. Tests drive it with class methods, and read what it was sent:

from typing import ClassVar

import frappe
from frappe import _

from acme_payments.services.acme.base import AcmeCheckout, AcmeRefund, BaseAcmeClient


class StubAcmeClient(BaseAcmeClient):
	checkouts: ClassVar[dict[str, AcmeCheckout]] = {}
	created_checkouts: ClassVar[list[dict]] = []
	refunds: ClassVar[list[dict]] = []
	next_refund_status: ClassVar[str] = "succeeded"

	@classmethod
	def reset(cls):
		cls.checkouts, cls.created_checkouts, cls.refunds = {}, [], []
		cls.next_refund_status = "succeeded"

	@classmethod
	def pay(cls, checkout_id: str):
		"""The shopper completes the hosted page."""
		cls.checkouts[checkout_id].status = "succeeded"

	def create_checkout(self, payload: dict) -> AcmeCheckout:
		self.created_checkouts.append(payload)
		checkout_id = f"chk_{frappe.generate_hash(length=12)}"
		self.checkouts[checkout_id] = AcmeCheckout(
			id=checkout_id, hosted_url=f"https://pay.acme.test/{checkout_id}", status="open"
		)
		return self.get_checkout(checkout_id)

	def get_checkout(self, checkout_id: str) -> AcmeCheckout:
		if checkout_id not in self.checkouts:
			frappe.throw(_("Acme Pay rejected the request"))
		return self.checkouts[checkout_id].model_copy()

	def create_refund(self, checkout_id: str, amount: int) -> AcmeRefund:
		self.refunds.append({"checkout_id": checkout_id, "amount": amount})
		return AcmeRefund(id=f"re_{frappe.generate_hash(length=8)}", status=self.next_refund_status, amount=amount)

• Make ids unique per call. A request's order_ref is unique, and rows can outlive a single test.
• Raise for an unknown id the same way the live client does.
• Record requests in Acme's own shape, so a test asserts on what Acme would really receive.

Step 5: Pick the client

Open acme_pay_gateway_settings.py. Extend PaymentGatewayBase alongside Document, and choose the client:

import frappe
from frappe.model.document import Document

from bwh_payments.base_class import CheckoutSession, PaymentGatewayBase, RefundResult, WebhookEvent

from acme_payments.services.acme.base import BaseAcmeClient
from acme_payments.services.acme.live import LiveAcmeClient
from acme_payments.services.acme.stub import StubAcmeClient


class AcmePayGatewaySettings(Document, PaymentGatewayBase):
	def get_gateway_name(self) -> str:
		return "Acme Pay"

	def get_client(self) -> BaseAcmeClient:
		return StubAcmeClient() if frappe.in_test else LiveAcmeClient(self.get_password("api_key"))

frappe.in_test is the only switch. A real site can never run on the stub.

PaymentGatewayBase is an abstract class. Until you've written all four methods below, Frappe can't load the settings and raises a TypeError naming the missing ones.

Step 6: Open a checkout session

	def create_session(
		self, amount, currency, reference=None, customer=None, return_url=None
	) -> CheckoutSession:
		currency = currency or self.currency
		validate_transaction_currency(currency)
		# With return_url, the app chose a page per payment; it replaces the settings URLs.
		success_url = return_url or get_localised_url(self.success_url)
		cancel_url = return_url or get_localised_url(self.cancelled_url)

		checkout = self.get_client().create_checkout(
			{
				"amount": to_minor_units(amount, currency),  # 49.99 USD -> 4999
				"currency": currency,
				"reference": reference,
				"email": (customer or {}).get("email") or None,
				"success_url": success_url,
				"cancel_url": cancel_url,
			}
		)
		return CheckoutSession(
			session_id=checkout.id,
			redirect_url=checkout.hosted_url,
			success_url=success_url,
			cancel_url=cancel_url,
			failure_url=cancel_url,
		)

Add these imports at the top of the module. Steps 7 to 9 use them too:

import hashlib
import hmac

from frappe import _

from bwh_payments.bwh_payments.utils import get_localised_url
from bwh_payments.currency import from_minor_units, to_minor_units, validate_transaction_currency

BWH Payments passes return_url only when the app set redirect_to on the payment. Send every outcome there, and add nothing to it.

Step 7: Report the payment status

ACME_STATUSES = {
	"succeeded": "Paid",
	"failed": "Not Paid",
	"cancelled": "Cancelled",
	"expired": "Expired",
}


	def get_payment_status(self, session_id: str) -> str:
		checkout = self.get_client().get_checkout(session_id)
		return ACME_STATUSES.get((checkout.status or "").strip().casefold(), "Pending")

Anything you don't recognise must map to Pending. Pending never ships goods and never cancels an order.

Step 8: Handle webhooks

Verify the signature in the controller, never in a client. A stub must never be able to accept a forged event.

	def handle_webhook(self, payload: bytes, headers: dict) -> WebhookEvent | None:
		secret = self.get_password("webhook_secret", raise_exception=False)
		if not secret:
			frappe.throw(_("Set the webhook secret in Acme Pay Gateway Settings"))

		signature = headers.get("X-Acme-Signature") or ""
		expected = hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest()
		if not signature.isascii() or not hmac.compare_digest(expected, signature):
			frappe.throw(_("Acme Pay webhook signature does not match"))

		event = frappe.parse_json(payload.decode())
		if event.get("type") != "checkout.succeeded":
			return None

		return WebhookEvent(session_id=event["data"]["id"], status="Paid", event_id=event["id"])

• Throw if verification fails. The gateway gets a 400.
• Return None for events you don't handle. The gateway gets a 200 and stops retrying.
• Return event_id, so a retried delivery is applied only once.

Webhooks and security has the full rules.

Step 9: Issue refunds

	def refund_payment(self, session_id: str, amount: float, currency: str | None = None) -> RefundResult:
		currency = currency or self.currency
		refund = self.get_client().create_refund(session_id, to_minor_units(amount, currency))
		if refund.status not in ("succeeded", "pending"):
			frappe.throw(_("Acme Pay refused the refund: {0}").format(refund.status))

		return RefundResult(
			refund_id=refund.id,
			status=refund.status,
			amount=from_minor_units(refund.amount, currency),
		)

BWH Payments has already checked the amount against what's left to refund. Throw if Acme Pay refuses, and the payment is left unchanged. After the money has moved, return what you have rather than raising: every RefundResult field is optional.

Step 10: Test it

Tests run your real controller against the stub. Nothing is patched:

# acme_payments/acme_payments/doctype/acme_pay_gateway_settings/test_acme_pay_gateway_settings.py
import frappe
from frappe.tests import IntegrationTestCase

from acme_payments.services.acme.stub import StubAcmeClient


class TestAcmePay(IntegrationTestCase):
	def setUp(self):
		StubAcmeClient.reset()
		settings = frappe.get_single("Acme Pay Gateway Settings")
		settings.api_key = "test_key"
		settings.webhook_secret = "test_secret"
		settings.save()

	def test_charges_in_minor_units(self):
		frappe.get_single("Acme Pay Gateway Settings").create_session(49.99, "USD", reference="GPR-1")
		self.assertEqual(StubAcmeClient.created_checkouts[-1]["amount"], 4999)

	def test_a_paid_checkout_is_paid(self):
		settings = frappe.get_single("Acme Pay Gateway Settings")
		session = settings.create_session(10, "USD")
		StubAcmeClient.pay(session.session_id)
		self.assertEqual(settings.get_payment_status(session.session_id), "Paid")

Test with stubs lists the cases every gateway should cover.

Step 11: Turn it on

1. Run bench --site your.site migrate.
2. Open Acme Pay Gateway Settings and fill in your sandbox keys and URLs.
3. Create a Payment Gateway Profile named Acme Pay, with Gateway Settings set to Acme Pay Gateway Settings, and tick Enabled. The profile won't save if your class doesn't implement the contract.
4. In Acme's dashboard, set the webhook URL to:

https://your.site/api/v2/method/bwh_payments.bwh_payments.webhook.handle?gateway=Acme%20Pay

5. Take a sandbox payment from the console:

from bwh_payments.api import create_payment

payment_request = create_payment("User", "Administrator", 10, "USD", "Acme Pay")
frappe.db.commit()  # bench console rolls back on exit, and the webhook must see this row
print(payment_request.order_url)  # open it, pay with a test card

After you pay, payment_request.reload().status should read Paid. If it doesn't, check the Integration Request and Error Log lists.

Next steps

• Service clients: why the controller and the client are split
• Gateway contract: every parameter, model and return value
• Razorpay source: a complete service client to compare against