BWH Shipping

BWH Shipping

Open in ChatGPT
Ask ChatGPT about this page
Open in Claude
Ask Claude about this page

Build a carrier

In this guide you connect a carrier called Acme Courier to BWH Shipping. By the end, checkout shows Acme's live rates, staff can book and cancel Acme Courier shipments from the desk, and Acme's webhooks move the status up the ladder. All of it is covered by tests that never call the real API.

What you'll learn

  • How to create a carrier settings DocType in your own app
  • How to implement the five methods of ShippingProviderBase
  • How to add optional capabilities, such as pickups
  • How to enable the carrier and test it without a live account

Requirements

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

Add BWH Shipping to your app's hooks.py:

required_apps = ["frappe/erpnext", "bwhtech/bwh_shipping"]

Step 1: Create the settings DocType

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

bench set-config -g developer_mode 1

Then create a new DocType in the desk:

  • Name: Acme Courier Shipping Settings
  • Module: a module in your app
  • Is Single: ✅ (BWH Shipping loads carriers with frappe.get_single)
  • Custom?: leave it unticked
  • Permissions: System Manager only

Add these fields:

Field Type Purpose
api_key Password Acme's secret API key
webhook_secret Password The secret Acme Courier signs webhooks with
pickup_address Link → Address Where parcels are collected. Checkout only asks for rates when this is set.
volumetric_divisor Int, default 5000 Read by BWH Shipping to compute billable weight
currency Link → Currency The currency Acme Courier quotes in

Step 2: Extend ShippingProviderBase

Open acme_courier_shipping_settings.py in your app:

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 bwh_shipping.base_class import ShippingProviderBase
from bwh_shipping.units import billable_weight, normalise_parcels, to_system_datetime

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


class AcmeCourierShippingSettings(Document, ShippingProviderBase):
	def get_provider_name(self) -> str:
		return "Acme Courier"

	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)

ShippingProviderBase is abstract: until all five required methods exist, Frappe can't load the settings and raises a TypeError naming the missing ones.

Step 3: Quote rates

get_rates runs at checkout, once per carrier. Return one row per service Acme Courier can offer:

	def get_rates(self, origin, destination, parcels, cod=False, declared_value=0.0) -> list[dict]:
		if not (origin.get("pincode") and destination.get("pincode")):
			return []

		parcels = normalise_parcels(parcels)
		quotes = self.request(
			"POST",
			"/rates",
			{
				"from": origin["pincode"],
				"to": destination["pincode"],
				"weight_kg": billable_weight(parcels, self.volumetric_divisor or 5000),
				"cod": cod,
			},
		)
		return [
			{
				"service_code": str(quote["service_id"]),
				"service_name": quote["name"],
				"carrier": "Acme Courier",
				"amount": quote["price"],
				"currency": self.currency,
				"transit_days": quote.get("days"),
				"cod_available": quote.get("cod", False),
			}
			for quote in quotes.get("services", [])
			if quote.get("price", 0) > 0
		]
  • Return [] when you can't quote. An empty list is a normal answer, and checkout falls back to the Shipping Rule or Backup Charge.
  • Normalise the parcels yourself. They arrive exactly as the checkout passed them.
  • Drop zero-price rows. A zero amount would show the shopper a free option.

Step 4: Book the shipment

create_shipment runs when staff click Book Shipment, or your code calls book():

	def create_shipment(self, shipment: dict) -> dict:
		label = self.request(
			"POST",
			"/shipments",
			{
				"reference": shipment["reference"],
				"service_id": shipment["service_code"],
				"from": shipment["origin"],
				"to": shipment["destination"],
				"parcels": normalise_parcels(shipment["parcels"]),
				"cod_amount": shipment["cod_amount"] if shipment["cod"] else 0,
			},
		)
		return {
			"order_ref": label["id"],
			"shipment_ref": label["id"],
			"awb": label["tracking_number"],
			"carrier": label.get("courier") or "Acme Courier",
			"label_url": label["label_url"],
			"cost_amount": label["price"],
			"cost_currency": self.currency,
			"status": label["status"],
		}

After this returns, the request's status becomes Ready To Ship, and these values are stored on it. Your status is stored as the carrier status.

Once you've paid for the label, never throw. If a follow-up call such as registering tracking fails, log it and return the booking anyway.

Step 5: Cancel a booking

	def cancel_shipment(self, order_ref, shipment_ref=None, awb=None) -> dict:
		response = self.request("POST", f"/shipments/{shipment_ref or order_ref}/cancel")
		return {"status": response.get("status", "cancelled"), "message": response.get("message", "")}

BWH Shipping only offers cancel while the request is Draft, Ready To Ship or Pickup Scheduled. Throw if Acme Courier refuses. On success the request becomes Cancelled.

Step 6: Report tracking

get_tracking runs when someone clicks Sync Status. Map Acme's statuses onto the ladder at module level, above the class:

ACME_STATUSES = {
	"created": "Ready To Ship",
	"picked_up": "In Transit",
	"in_transit": "In Transit",
	"out_for_delivery": "Out For Delivery",
	"failed_attempt": "Undelivered",
	"returned": "RTO",
	"delivered": "Delivered",
	"lost": "Lost",
}

Then add the method to the class:

	def get_tracking(self, awb, shipment_ref=None, tracking_ref=None) -> dict:
		tracking = self.request("GET", f"/tracking/{awb}")
		return {
			"status": ACME_STATUSES.get(tracking.get("status")),
			"provider_status": tracking.get("status"),
			"events": [
				{
					"timestamp": to_system_datetime(scan["time"]),
					"status": scan["status"],
					"location": scan.get("location"),
					"message": scan.get("description"),
				}
				for scan in tracking.get("scans", [])
			],
		}
  • status must be a ladder status, or None for anything you don't recognise. None leaves the status alone.
  • Pass timestamps through to_system_datetime. Timestamps with a timezone offset can't be stored as they are.
  • Scans are added to the request's tracking events, and repeats are skipped.

Step 7: Handle webhooks

Acme Courier posts to one shared BWH Shipping endpoint, which hands you the raw body and headers:

	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 Courier Shipping 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 Courier webhook signature does not match"))

		event = frappe.parse_json(payload.decode())
		status = ACME_STATUSES.get(event["status"])
		if not status:
			return {}

		return {
			"awb": event["tracking_number"],
			"status": status,
			"provider_status": event["status"],
			"event_id": event["event_id"],
			"events": [],
		}

Throw if the delivery isn't genuine. Return {} for events that don't map to a ladder status, because a result without a status is dropped. Webhooks and security has the full rules.

Step 8: Add optional capabilities

Implement only what Acme Courier really supports. supports() decides whether BWH Shipping runs the action or refuses it:

Method Capability Adds
schedule_pickup(shipment_ref, pickup_date=None) pickup Schedule Pickup works
generate_manifest(shipment_refs) manifest Generate Manifest works
resume_booking(order_ref, shipment_ref, service_code=None) resume Resume Booking works after a PartialBookingError
get_service_choices() service_choices Your code can list the carrier's services

supports("pickup") is True only when your class overrides the method. The Schedule Pickup and Generate Manifest buttons show on every booked shipment, and for a carrier without the capability they fail with a clear message. Never override a method just to raise an error, because supports() would then report True. The Import from Carrier button in the Shipping Service list only works for AfterShip.

Step 9: Test it

Replace the request helpers in your module with a fake, so your real class runs without the network:

# acme_shipping/tests/test_acme_courier.py: adjust the import to your app and module
from unittest.mock import patch

import frappe
from frappe.tests import IntegrationTestCase

from acme_shipping.acme_shipping.doctype.acme_courier_shipping_settings import (
	acme_courier_shipping_settings,
)


class FakeAcme:
	sent = []

	@classmethod
	def post(cls, url, json=None, headers=None, **kwargs):
		cls.sent.append(json)
		return {"services": [{"service_id": 7, "name": "Express", "price": 120, "days": 2}]}


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

		settings = frappe.get_single("Acme Courier Shipping Settings")
		settings.api_key = "test_key"  # get_password throws when a Password field is empty
		settings.currency = "INR"
		settings.save()

	def test_quotes_on_billable_weight(self):
		rates = frappe.get_single("Acme Courier Shipping Settings").get_rates(
			{"pincode": "110001"},
			{"pincode": "400001"},
			[{"length": 50, "width": 40, "height": 30, "weight": 2}],
		)
		self.assertEqual(FakeAcme.sent[-1]["weight_kg"], 12)  # 50×40×30 / 5000 beats 2 kg
		self.assertEqual(rates[0]["service_code"], "7")

Run them on a test site with ERPNext:

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

Cover at least:

  • rates for a normal and a bulky parcel, and [] for a missing pincode
  • each Acme Courier status maps to the right ladder status, and an unknown one maps to None
  • a correctly signed webhook returns the AWB and status
  • a bad signature, a missing header and a missing secret all throw

Step 10: Turn it on

  1. Run bench --site your.site migrate.

  2. Open Acme Courier Shipping Settings and fill in the key, secret, Pickup Address and currency.

  3. Create a Shipping Provider Profile:

    • Name: Acme Courier
    • Provider Settings: Acme Courier Shipping Settings
    • Enabled: ✅

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

  4. Create a Shipping Service for each Acme Courier service you sell, with its Service Code.

  5. In Acme's dashboard, set the webhook URL to:

    https://your.site/api/v2/method/bwh_shipping.bwh_shipping.webhook.handle?provider=Acme%20Courier
    
  6. Create a Shipping Request for a test order, and click Book Shipment.

Next steps

Last updated 3 weeks ago
Was this helpful?
Thanks!