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. All of it is covered by tests that never call the real API.

What you'll learn

• How to create a gateway settings DocType in your own app
• How to implement the four methods of PaymentGatewayBase
• How to verify webhooks safely
• How to enable the gateway and test it without a live account

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"]

Step 1: Create the settings DocType

The settings DocType stores Acme's credentials and return URLs. Its controller class is the integration.

Turn on developer mode first, so Frappe writes the DocType and its Python controller into your app:

bench set-config -g developer_mode 1

Then create a new DocType in the desk:

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

Add these fields:

Step 2: Extend PaymentGatewayBase

Open acme_pay_gateway_settings.py in your app and extend PaymentGatewayBase alongside Document:

import hashlib
import hmac

import frappe
from frappe import _
from frappe.integrations.utils import make_get_request, make_post_request
from frappe.model.document import Document
from frappe.utils.data import flt

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

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


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

	def request(self, method, path: str, payload: dict | None = None) -> dict:
		headers = {"Authorization": f"Bearer {self.get_password('api_key')}"}
		if method == "GET":
			return make_get_request(f"{ACME_API}{path}", headers=headers)
		return make_post_request(f"{ACME_API}{path}", json=payload, headers=headers)

PaymentGatewayBase is an abstract class. Until you've written all four methods, Frappe can't load the settings and raises a TypeError naming the missing ones. That's deliberate: an incomplete gateway fails as soon as it's opened, not halfway through a checkout.

Step 3: Open a checkout session

create_session runs when your app creates a Gateway Payment Request. Ask Acme Pay for a hosted checkout, then return its id and URL:

	def create_session(self, amount, currency, reference=None, customer=None) -> dict:
		currency = currency or self.currency
		validate_transaction_currency(currency)

		checkout = self.request(
			"POST",
			"/checkouts",
			{
				"amount": to_minor_units(amount, currency),  # 49.99 USD -> 4999
				"currency": currency,
				"reference": reference,
				"email": (customer or {}).get("email") or None,
				"success_url": get_localised_url(self.success_url),
				"cancel_url": get_localised_url(self.cancelled_url),
			},
		)
		return {
			"session_id": checkout["id"],
			"redirect_url": checkout["hosted_url"],
			"success_url": get_localised_url(self.success_url),
			"cancel_url": get_localised_url(self.cancelled_url),
			"failure_url": get_localised_url(self.cancelled_url),
		}

Leave out empty customer fields. Many gateways return a 400 for a blank email or phone number, which breaks guest checkout.

Step 4: Report the payment status

get_payment_status is how BWH Payments checks a payment without waiting for a webhook, for example when the shopper returns to your site. Read the status from Acme Pay and map it onto BWH Payments' statuses:

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

	def get_payment_status(self, session_id: str) -> str:
		checkout = self.request("GET", f"/checkouts/{session_id}")
		return ACME_STATUSES.get(checkout.get("status"), "Pending")

Anything you don't recognise must map to Pending. Pending never ships goods and never cancels an order, so an unexpected status is harmless.

Step 5: Handle webhooks

Acme Pay posts to one shared BWH Payments endpoint, which passes the raw body and headers to handle_webhook. Verify the signature first. Then return what happened, or {} to ignore the delivery:

	def handle_webhook(self, payload: bytes, headers: dict) -> dict:
		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 {}

		return {
			"session_id": event["data"]["id"],
			"status": "Paid",
			"event_id": event["id"],
		}

• Throw if verification fails. The caller gets a 400.
• Return {} for events you don't handle. The caller gets a 200, so Acme Pay stops retrying.
• Return event_id so a retried delivery is only applied once.

Webhooks and security has the full rules, including what to do if Acme Pay doesn't sign its webhooks.

Step 6: Issue refunds

refund_payment is called with an amount that BWH Payments has already checked against what's left to refund:

	def refund_payment(self, session_id: str, amount: float, currency: str | None = None) -> dict:
		currency = currency or self.currency
		refund = self.request(
			"POST", f"/checkouts/{session_id}/refunds", {"amount": to_minor_units(amount, currency)}
		)
		if refund.get("status") not in ("succeeded", "pending"):
			frappe.throw(_("Acme Pay refused the refund: {0}").format(refund.get("status")))

		return {
			"refund_id": refund["id"],
			"status": refund["status"],
			"amount": flt(from_minor_units(refund["amount"], currency)),
		}

Throw if Acme Pay refuses, and BWH Payments leaves the payment unchanged. BWH Payments stores refund_id for reconciliation. status and amount are informational: the refunded total is increased by the amount it asked for. Convert charges and refunds the same way, so what Acme Pay refunds is what you requested.

Step 7: Let shoppers cancel (optional)

If Acme Pay can cancel an open checkout, override cancel_session. It's called from payment_request.release_if_unpaid(), which your app calls when a shopper abandons a checkout, for example to let them change their order:

	def cancel_session(self, session_id: str) -> bool:
		try:
			checkout = self.request("POST", f"/checkouts/{session_id}/cancel")
		except Exception:
			return False
		return checkout.get("status") == "cancelled"

Return True only when Acme Pay confirms the checkout can no longer be paid. If Acme Pay has no cancel endpoint, don't override it. The default returns False, and release_if_unpaid() then reports that the payment is still open.

Step 8: Test it

Your tests should run your real class against a fake Acme Pay. Replace the two request helpers in your module with a fake that records what was sent:

# acme_payments/tests/test_acme_pay.py: adjust the import below to your app and module
from unittest.mock import patch

import frappe
from frappe.tests import IntegrationTestCase

from acme_payments.acme_payments.doctype.acme_pay_gateway_settings import acme_pay_gateway_settings


class FakeAcme:
	sent = []

	@classmethod
	def post(cls, url, json=None, headers=None, **kwargs):
		cls.sent.append(json)
		return {"id": "chk_123", "hosted_url": "https://pay.acme.example/chk_123"}


class TestAcmePay(IntegrationTestCase):
	def setUp(self):
		FakeAcme.sent = []
		transport = patch.object(acme_pay_gateway_settings, "make_post_request", FakeAcme.post)
		transport.start()
		self.addCleanup(transport.stop)

		settings = frappe.get_single("Acme Pay Gateway Settings")
		settings.api_key = "test_key"  # get_password throws when a Password field is empty
		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(FakeAcme.sent[-1]["amount"], 4999)

Run the tests:

bench --site test_site set-config allow_tests true
bench --site test_site run-tests --app acme_payments

Cover at least these cases:

• the exact amount sent for USD (2 decimals), KWD (3) and JPY (0)
• each Acme Pay status maps correctly, and an unknown one maps to Pending
• a correctly signed webhook returns Paid
• a bad signature, a missing header and a missing secret all throw
• a partial refund followed by the rest leaves the request Refunded

For a complete example, including a fake that keeps state and signs real webhook payloads, see BWH Payments' own Razorpay fake and its tests.

Step 9: 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:

   • Name: Acme Pay, which is what apps pass as gateway
   • Gateway Settings: Acme Pay Gateway Settings
   • Enabled: ✅

   The profile won't save if your class doesn't extend PaymentGatewayBase.

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 real sandbox payment from the console:

payment_request = frappe.get_doc(
	{
		"doctype": "Gateway Payment Request",
		"gateway": "Acme Pay",
		"amount": 10,
		"currency_code": "USD",
		"ref_doctype": "User",
		"ref_docname": "Administrator",
	}
).insert()
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

• Gateway reference: every parameter and return value in detail
• Webhooks and security: gateways that don't sign, replays, and debugging
• Stripe controller source: a complete, production gateway to compare against