Examples
Whole integrations, written out. Copy one into a new script, give it the trigger it names, put its keys in Secrets, and change whatever you like — from then on it is your code.
On this page
Who posted it
A script's address is a long secret token, and can also check a shared-secret
signature. Stripe, Shopify and GitHub sign their own way, which an address does not
check — for an event that moves money, confirm it against the sender's API before
acting on it.
Stripe payment → row and a Slack message #
Takes payment_intent.succeeded, writes the payment to a table with db.sql, and posts a line to Slack.
| Runs on | an address another application posts to |
| Table |
payments
(stripe_id text, amount real, currency text, email email, status text, refunded real)
|
def main(event):
# Stripe posts many kinds of event to one address; this handles one and ignores the
# rest. Adding a second is another `if`, not another address.
if event["event"] != "payment_intent.succeeded":
return
intent = event["payload"]["data"]["object"]
amount = intent["amount"] / 100.0
currency = intent["currency"].upper()
# Stripe delivers an event again when it is not sure the first one landed. Looking
# the payment up by its Stripe id first is what makes the second delivery update the
# row instead of adding another one for the same money.
if db["shop"].get("payments", stripe_id=intent["id"]):
db["shop"].sql(
"UPDATE payments SET status = ? WHERE stripe_id = ?",
["succeeded", intent["id"]],
)
else:
db["shop"].sql(
"INSERT INTO payments (stripe_id, amount, currency, email, status) "
"VALUES (?, ?, ?, ?, ?)",
[intent["id"], amount, currency, intent.get("receipt_email") or "", "succeeded"],
)
# The payment is written by now; the message is a courtesy. If Slack is down the run
# should still count as handled — a failed run is retried, and the retry would find
# nothing left to do but try Slack again.
try:
http(
secrets["SLACK_WEBHOOK_URL"],
json={"text": "Payment of %.2f %s" % (amount, currency)},
timeout=10,
)
except Exception as exc:
print("could not post to Slack:", exc)
return {"payment": intent["id"], "amount": amount, "currency": currency}
Stripe refund → update the payment #
Takes charge.refunded and updates the row the payment template wrote, rather than adding a second one for the same money.
| Runs on | an address another application posts to |
| Table |
payments
(stripe_id text, amount real, currency text, email email, status text, refunded real)
|
def main(event):
if event["event"] != "charge.refunded":
return
charge = event["payload"]["data"]["object"]
refunded = charge["amount_refunded"] / 100.0
status = "refunded" if charge.get("refunded") else "partly refunded"
# The same row the payment example wrote, found by the same column. A refund names
# the charge, so the payment intent is where the two meet.
stripe_id = charge.get("payment_intent") or charge["id"]
if db["shop"].get("payments", stripe_id=stripe_id):
db["shop"].sql(
"UPDATE payments SET refunded = ?, status = ? WHERE stripe_id = ?",
[refunded, status, stripe_id],
)
else:
# A refund for a payment this table never saw — one from before the script
# existed. Keep it anyway, so the money is accounted for.
db["shop"].sql(
"INSERT INTO payments (stripe_id, amount, currency, refunded, status) "
"VALUES (?, ?, ?, ?, ?)",
[stripe_id, charge["amount"] / 100.0, charge["currency"].upper(), refunded, status],
)
Stripe payouts → a table, every morning #
The other direction: nothing posts here. A clock runs it at 06:00 UTC and it calls Stripe with the official SDK to keep a payouts table current.
| Runs on | a schedule, 0 6 * * * (UTC) |
| Packages | stripe |
| Table |
payouts
(stripe_id text, amount real, currency text, status text, arrives_on integer)
|
import stripe
def main(event):
# Nothing comes in on a schedule: this is the code's turn to go and look.
stripe.api_key = secrets["STRIPE_KEY"]
# The latest twenty, not the whole history. A run may make 50 statements and each
# payout here is a lookup and a write; a morning never brings twenty new ones.
for payout in stripe.Payout.list(limit=20).data:
values = [
payout.amount / 100.0,
payout.currency.upper(),
payout.status,
payout.arrival_date,
]
if db["shop"].get("payouts", stripe_id=payout.id):
db["shop"].sql(
"UPDATE payouts SET amount = ?, currency = ?, status = ?, arrives_on = ? "
"WHERE stripe_id = ?",
values + [payout.id],
)
else:
db["shop"].sql(
"INSERT INTO payouts (amount, currency, status, arrives_on, stripe_id) "
"VALUES (?, ?, ?, ?, ?)",
values + [payout.id],
)
Shopify order → rows #
Takes orders/create and orders/updated, writes the order and each of its line items.
| Runs on | an address another application posts to |
| Table |
shopify_orders
(shop_order_id text, number integer, email email, total text, currency text, status text, placed_at text)
|
| Table |
shopify_items
(line_id text, shop_order_id text, title text, quantity integer, price text)
|
def main(event):
# Shopify names its event in the X-Shopify-Topic header, not in the body. This
# application reads that header, so `orders/create` arrives as it was sent.
if event["event"] not in ("orders/create", "orders/updated"):
return
order = event["payload"]
customer = order.get("customer") or {}
order_id = str(order["id"])
# Money arrives as a string — "42.50", not 42.5 — because a float cannot hold a
# price exactly. Keep it as text and convert only where you need arithmetic.
values = [
order.get("order_number"),
order.get("email") or customer.get("email") or "",
order.get("total_price") or "0",
order.get("currency") or "",
order.get("financial_status") or "",
order.get("created_at") or "",
]
# orders/updated is the same order again, so it is found by its Shopify id and
# updated rather than added a second time.
if db["shop"].get("shopify_orders", shop_order_id=order_id):
db["shop"].sql(
"UPDATE shopify_orders SET number = ?, email = ?, total = ?, currency = ?, "
"status = ?, placed_at = ? WHERE shop_order_id = ?",
values + [order_id],
)
else:
db["shop"].sql(
"INSERT INTO shopify_orders "
"(number, email, total, currency, status, placed_at, shop_order_id) "
"VALUES (?, ?, ?, ?, ?, ?, ?)",
values + [order_id],
)
# The lines are replaced as a set: an edited order can lose a line as well as gain
# one, and one INSERT for all of them keeps a fifty-line order to two statements.
db["shop"].sql("DELETE FROM shopify_items WHERE shop_order_id = ?", [order_id])
items = order.get("line_items") or []
if items:
params = []
for item in items:
params += [
str(item["id"]),
order_id,
item.get("title") or "",
item.get("quantity") or 0,
item.get("price") or "0",
]
# Only the placeholders are joined into the statement; the values stay in params.
db["shop"].sql(
"INSERT INTO shopify_items (line_id, shop_order_id, title, quantity, price) "
"VALUES " + ", ".join(["(?, ?, ?, ?, ?)"] * len(items)),
params,
)
return {"order": order_id, "lines": len(items)}
Stripe payout → QuickBooks deposit #
Every morning, each Stripe transfer becomes one deposit in QuickBooks, with the fee as its own negative line so it reconciles against the bank.
| Runs on | a schedule, 0 7 * * * (UTC) |
| Packages | stripe |
| Table |
qbo_deposits
(payout_id text, deposit_id text, gross real, fees real, net real, currency text, arrives_on integer, note text)
|
import json
import urllib.error
import urllib.parse
import urllib.request
import stripe
# ── Three account ids, filled in once ─────────────────────────────────────────
# QuickBooks books money into accounts by id, never by name, and there is no way to
# guess which of yours is which. Leave these blank and the first run prints the list
# with the ids beside it.
#
# BANK — where Stripe's transfer actually arrives.
# FEES — where Stripe's cut goes: "Bank Charges", or an expense account of your
# own called "Stripe fees".
# INCOME — what the sales themselves are booked to.
BANK_ACCOUNT_ID = ""
FEES_ACCOUNT_ID = ""
INCOME_ACCOUNT_ID = ""
def main(event):
if not (BANK_ACCOUNT_ID and FEES_ACCOUNT_ID and INCOME_ACCOUNT_ID):
for account in qbo_query("select Id, Name, AccountType from Account", "Account"):
print(account["Id"], account["AccountType"], account["Name"])
raise RuntimeError("fill in the three account ids at the top of this script")
stripe.api_key = secrets["STRIPE_KEY"]
# The latest twenty paid payouts. Each is a lookup here and maybe a write, and a run
# may make 50 statements; a morning never brings twenty new ones.
for payout in stripe.Payout.list(status="paid", limit=20).data:
# A Deposit takes no idempotency key, so nothing on QuickBooks' side stops a
# second run posting the same money again. This table is what stops it.
if db["shop"].get("qbo_deposits", payout_id=payout.id):
continue
deposit(payout)
# What counts as a sale. Everything else in a payout — Stripe's own monthly billing, a
# dispute, a topup — came out of the same transfer but is not income, and booking it to
# the income account would quietly overstate sales and understate costs by the same
# amount, which is the kind of error that balances and is still wrong.
SALES = ("charge", "payment", "refund", "payment_refund", "adjustment")
def deposit(payout):
# Every balance transaction in the payout except the payout itself. Their amounts
# less their fees come to exactly payout.amount, and that identity is what makes
# the deposit reconcile against the line on the bank statement.
sales = 0
fees = 0
for item in stripe.BalanceTransaction.list(payout=payout.id, limit=100).auto_paging_iter():
if item.type == "payout":
continue
if item.type in SALES:
# A refund arrives here as a negative amount, so this one sum covers both
# directions and there is no separate refund case to get wrong.
sales += item.amount
else:
fees -= item.amount
fees += item.fee
if sales - fees <= 0:
# Stripe debits the bank when a day's refunds beat its sales. That is a
# withdrawal, not a deposit, and QuickBooks will not accept it as one.
db["shop"].sql(
"INSERT INTO qbo_deposits (payout_id, net, currency, arrives_on, note) "
"VALUES (?, ?, ?, ?, ?)",
[
payout.id,
(sales - fees) / 100.0,
payout.currency.upper(),
payout.arrival_date,
"negative payout — enter this one by hand",
],
)
return
created = qbo("deposit", {
"DepositToAccountRef": {"value": BANK_ACCOUNT_ID},
# Harmless when it matches the company's own currency, and required when it
# does not — in which case multi-currency has to be on in QuickBooks.
"CurrencyRef": {"value": payout.currency.upper()},
"PrivateNote": "Stripe payout %s" % payout.id,
"Line": [
{
"Amount": round(sales / 100.0, 2),
"DetailType": "DepositLineDetail",
"Description": "Stripe sales",
"DepositLineDetail": {"AccountRef": {"value": INCOME_ACCOUNT_ID}},
},
{
# Negative on purpose. The fee came out before the money arrived, and
# a negative line is how a deposit says so without a second
# transaction that would then have to be matched to this one.
"Amount": round(-fees / 100.0, 2),
"DetailType": "DepositLineDetail",
"Description": "Stripe fees",
"DepositLineDetail": {"AccountRef": {"value": FEES_ACCOUNT_ID}},
},
],
})["Deposit"]
# Written straight after QuickBooks said yes, so the next run skips this payout.
db["shop"].sql(
"INSERT INTO qbo_deposits "
"(payout_id, deposit_id, gross, fees, net, currency, arrives_on, note) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
[
payout.id,
created["Id"],
round(sales / 100.0, 2),
round(fees / 100.0, 2),
round((sales - fees) / 100.0, 2),
payout.currency.upper(),
payout.arrival_date,
"",
],
)
def qbo(path, body=None, query=None):
params = {"minorversion": "75"}
if query:
params["query"] = query
request = urllib.request.Request(
"https://%s/v3/company/%s/%s?%s"
% (
secrets["QUICKBOOKS_API"],
secrets["QUICKBOOKS_REALM"],
path,
urllib.parse.urlencode(params),
),
data=json.dumps(body).encode() if body is not None else None,
headers={
# A fresh access token every run. The refresh that produced it happened
# on the way in, and this script never learns that OAuth is involved.
"authorization": "Bearer " + secrets["QUICKBOOKS_TOKEN"],
"content-type": "application/json",
"accept": "application/json",
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as exc:
# QuickBooks explains itself in the body, and urllib throws the body away
# unless you read it off the exception. Without this line every mistake is
# "HTTP Error 400: Bad Request" with no reason attached.
raise RuntimeError("QuickBooks refused: %s" % exc.read().decode("utf-8", "replace"))
def qbo_query(statement, kind):
answer = qbo("query", query=statement)
return answer.get("QueryResponse", {}).get(kind, [])
Shopify order → QuickBooks invoice #
Takes orders/create, finds or makes the customer, maps each line by SKU, and posts an invoice under the Shopify order number.
| Runs on | an address another application posts to |
| Table |
qbo_invoices
(shop_order_id text, invoice_id text, doc_number text, invoice_total real, shopify_total text, shopify_tax text, currency text)
|
import json
import urllib.error
import urllib.parse
import urllib.request
# ── Two item ids, filled in once ──────────────────────────────────────────────
# An invoice line has to point at a QuickBooks item, and Shopify's products are not
# QuickBooks items. Each line is looked up by SKU first; DEFAULT_ITEM_ID is the
# fallback when there is no match, and SHIPPING_ITEM_ID is for the shipping line.
# Leave them blank and the first run prints your items with their ids.
DEFAULT_ITEM_ID = ""
SHIPPING_ITEM_ID = ""
def main(event):
# Shopify names its event in the X-Shopify-Topic header; this application reads
# that header, so the topic arrives as it was sent.
if event["event"] not in ("orders/create", "orders/paid"):
return
if not DEFAULT_ITEM_ID:
for item in qbo_query("select Id, Name, Sku from Item", "Item"):
print(item["Id"], item.get("Sku") or "-", item["Name"])
raise RuntimeError("fill in the item ids at the top of this script")
order = event["payload"]
# Shopify retries a delivery it is not sure landed. An invoice posted twice is a
# conversation with an accountant, so this check comes before anything else.
if db["shop"].get("qbo_invoices", shop_order_id=str(order["id"])):
return
invoice(order)
def invoice(order):
lines = [line_for(item) for item in order.get("line_items") or []]
shipping = sum(float(line.get("price") or 0) for line in order.get("shipping_lines") or [])
if shipping > 0:
lines.append({
"Amount": round(shipping, 2),
"DetailType": "SalesItemLineDetail",
"Description": "Shipping",
"SalesItemLineDetail": {"ItemRef": {"value": SHIPPING_ITEM_ID}},
})
# Tax is deliberately not mapped. Shopify has already worked out what it charged
# and QuickBooks will work out its own from the customer and the company's tax
# settings; which of the two is right is a question about your jurisdiction, not
# a line of code. Both totals go in the row below so any gap is visible.
created = qbo("invoice", {
"CustomerRef": {"value": customer_for(order)},
"CurrencyRef": {"value": order.get("currency") or "USD"},
# Shopify's own number, so the two systems can be lined up by eye.
"DocNumber": str(order.get("order_number") or order["id"])[:21],
"PrivateNote": "Shopify order %s" % order["id"],
"Line": lines,
})["Invoice"]
# Written straight after QuickBooks said yes, so a redelivery stops at the check
# in main.
db["shop"].sql(
"INSERT INTO qbo_invoices (shop_order_id, invoice_id, doc_number, invoice_total, "
"shopify_total, shopify_tax, currency) VALUES (?, ?, ?, ?, ?, ?, ?)",
[
str(order["id"]),
created["Id"],
created.get("DocNumber") or "",
created.get("TotalAmt") or 0,
order.get("total_price") or "0",
order.get("total_tax") or "0",
order.get("currency") or "",
],
)
def customer_for(order):
person = order.get("customer") or {}
email = (order.get("email") or person.get("email") or "").strip()
if email:
found = qbo_query(
"select Id from Customer where PrimaryEmailAddr = '%s'" % escape(email),
"Customer",
)
if found:
return found[0]["Id"]
first = person.get("first_name") or ""
last = person.get("last_name") or ""
name = ("%s %s" % (first, last)).strip() or email or "Shopify customer"
body = {"DisplayName": name[:100]}
if email:
body["PrimaryEmailAddr"] = {"Address": email}
if first:
body["GivenName"] = first
if last:
body["FamilyName"] = last
try:
return qbo("customer", body)["Customer"]["Id"]
except RuntimeError as exc:
# DisplayName is unique in QuickBooks, so a second "John Smith" is refused
# outright. The one already there is the one this wanted.
if "Duplicate Name Exists" not in str(exc):
raise
found = qbo_query(
"select Id from Customer where DisplayName = '%s'" % escape(name[:100]),
"Customer",
)
if not found:
raise
return found[0]["Id"]
def line_for(item):
sku = (item.get("sku") or "").strip()
item_id = DEFAULT_ITEM_ID
if sku:
found = qbo_query("select Id from Item where Sku = '%s'" % escape(sku), "Item")
if found:
item_id = found[0]["Id"]
quantity = item.get("quantity") or 1
price = float(item.get("price") or 0)
return {
"Amount": round(price * quantity, 2),
"DetailType": "SalesItemLineDetail",
"Description": item.get("title") or "",
"SalesItemLineDetail": {
"ItemRef": {"value": item_id},
"Qty": quantity,
"UnitPrice": price,
},
}
def escape(value):
# QuickBooks' query language looks like SQL and is not: a quote inside a string is
# escaped with a backslash, and doubling it — the SQL habit — silently matches
# nothing instead of raising.
return value.replace("\\", "\\\\").replace("'", "\\'")
def qbo(path, body=None, query=None):
params = {"minorversion": "75"}
if query:
params["query"] = query
request = urllib.request.Request(
"https://%s/v3/company/%s/%s?%s"
% (
secrets["QUICKBOOKS_API"],
secrets["QUICKBOOKS_REALM"],
path,
urllib.parse.urlencode(params),
),
data=json.dumps(body).encode() if body is not None else None,
headers={
"authorization": "Bearer " + secrets["QUICKBOOKS_TOKEN"],
"content-type": "application/json",
"accept": "application/json",
},
)
try:
with urllib.request.urlopen(request, timeout=30) as response:
return json.load(response)
except urllib.error.HTTPError as exc:
# The body is where QuickBooks says what was wrong, and urllib hides it on
# the exception rather than in the message.
raise RuntimeError("QuickBooks refused: %s" % exc.read().decode("utf-8", "replace"))
def qbo_query(statement, kind):
answer = qbo("query", query=statement)
return answer.get("QueryResponse", {}).get(kind, [])