Scripts

        import requests

def main(event):
    # Nothing comes in: a scheduled run is the code's turn to go and look.
    orders = requests.get(
        "https://api.example.com/orders?since=yesterday&limit=40",
        headers={"Authorization": "Bearer " + secrets["SHOP_KEY"]},
    ).json()

    for order in orders:
        # PostgreSQL; needs a unique index on orders.code.
        db["shop"].sql(
            "INSERT INTO orders (code, total) VALUES (?, ?) "
            "ON CONFLICT (code) DO UPDATE SET total = excluded.total",
            [order["id"], order["total"]],
        )

      

def main(event):
    order = event["payload"]

    # A read is a list of dicts.
    seen = db["shop"].sql("SELECT id, status FROM orders WHERE code = ?", [order["id"]])

    # A write is the same call; .rowcount says how many rows it changed.
    if seen:
        changed = db["shop"].sql("UPDATE orders SET total = ? WHERE id = ?",
                         [order["total"], seen[0]["id"]])
        print("updated", changed.rowcount)
    else:
        db["shop"].sql("INSERT INTO orders (code, total) VALUES (?, ?)",
               [order["id"], order["total"]])

    db["shop"].sql("DELETE FROM carts WHERE email = ?", [order["email"]])

export default async function main(event: CmsEvent) {
  const order = event.payload;

  const changed = await db.source("shop").sql(
    "UPDATE orders SET status = ? WHERE code = ?",
    ["paid", order.id],
  );
  console.log("marked paid:", changed.rowCount);

  const paid = await db.source("shop").sql("SELECT code, total FROM orders WHERE status = ?", ["paid"]);
  return { paid: paid.length };
}

row  = db["shop"].get("orders", code="A-1")        # one row, or None
rows = db["shop"].all("orders", order_by="id")     # every row, up to the cap

def main(event):
    order = db["shop"].get("orders", code=event["payload"]["code"])

    # db["crm"] is the data source called crm: the same calls, another server.
    db["crm"].sql("UPDATE contacts SET last_order = ? WHERE email = ?",
                  [order["code"], order["email"]])

def main(event):
    order = event["payload"]

    # Both statements commit when the block ends, or neither does: an exception
    # inside it rolls back everything the block wrote.
    with db["shop"].transaction():
        db["shop"].sql("UPDATE stock SET qty = qty - ? WHERE sku = ?",
               [order["qty"], order["sku"]])
        db["shop"].sql("INSERT INTO shipments (code, sku) VALUES (?, ?)",
               [order["id"], order["sku"]])

await db.source("shop").transaction(async () => {
  await db.source("shop").sql("UPDATE stock SET qty = qty - ? WHERE sku = ?", [order.qty, order.sku]);
  await db.source("shop").sql("INSERT INTO shipments (code, sku) VALUES (?, ?)", [order.id, order.sku]);
});

# Postgres and MariaDB
added = db["shop"].sql("INSERT INTO orders (code, total) VALUES (?, ?) RETURNING id",
               ["A-9", 120])
new_id = added[0]["id"]

# SQL Server
added = db["shop"].sql("INSERT INTO orders (code, total) OUTPUT INSERTED.id VALUES (?, ?)",
               ["A-9", 120])

import sqlalchemy as sa

def main(event):
    engine = sa.create_engine("10lines://shop")

    # engine.begin() commits when the block ends, and rolls back if it raises.
    with engine.begin() as c:
        orders = sa.Table("orders", sa.MetaData(), autoload_with=c)
        late = c.execute(
            sa.select(orders.c.id, orders.c.code).where(orders.c.status == "new")
        ).mappings().all()

        c.execute(
            sa.update(orders)
            .where(orders.c.id.in_([row["id"] for row in late]))
            .values(status="late")
        )

    return [row["code"] for row in late]

import sqlalchemy as sa
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    code: Mapped[str]
    total: Mapped[int]

def main(event):
    # 10lines://crm is the data source called crm, on its own database's dialect.
    engine = sa.create_engine("10lines://shop")

    with Session(engine) as session:
        order = Order(code=event["payload"]["id"], total=event["payload"]["total"])
        session.add(order)
        session.commit()
        return {"id": order.id}

https://10lines.io/hooks/<project>/<token>

curl -X POST https://10lines.io/hooks/7/whin_Kd3qP1xY.../order.paid \
  -H 'content-type: application/json' \
  -d '{"id": "A-1001", "customer": "Ada Lovelace", "total": 42.5}'

HTTP/1.1 202 Accepted
content-type: application/json

{"ok": true, "status": "queued", "event": "order.paid", "id": 913}

# Python — Flask
import hmac, hashlib, time
from flask import Flask, request

SECRET = b"the value of SYNC_SECRET"
app = Flask(__name__)


@app.post("/hooks/10lines")
def receive():
    body = request.get_data()
    timestamp = request.headers.get("x-10lines-timestamp", "")
    signature = request.headers.get("x-10lines-signature", "")

    # Refuse anything old, so a request somebody captured cannot be replayed.
    if abs(time.time() - int(timestamp)) > 300:
        return "stale", 400

    expected = "sha256=" + hmac.new(
        SECRET, f"{timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        return "bad signature", 400

    return "", 200

// Node — Express
const crypto = require("crypto")
const express = require("express")

const SECRET = "the value of SYNC_SECRET"
const app = express()

// The raw body is what was signed, so keep it before anything parses it.
app.post("/hooks/10lines", express.raw({type: "application/json"}), (req, res) => {
  const timestamp = req.get("x-10lines-timestamp")
  const signature = req.get("x-10lines-signature") || ""

  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) {
    return res.status(400).send("stale")
  }

  const expected = "sha256=" + crypto
    .createHmac("sha256", SECRET)
    .update(`${timestamp}.`)
    .update(req.body)
    .digest("hex")

  const ok = expected.length === signature.length &&
    crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature))

  if (!ok) { return res.status(400).send("bad signature") }

  res.sendStatus(200)
})

def main(event):
    db["shop"].sql("INSERT INTO posts (title) VALUES (?)", [event["payload"]["name"]])

export default async function main(event: CmsEvent) {
  await db.source("shop").sql("INSERT INTO posts (title) VALUES (?)", [event.payload.name]);
}

{
  "id": 913,                              # this event's id in the log
  "event": "order.paid",                  # the name it arrived under
  "payload": {"id": "A-1001", ...},       # the body that was posted
  "headers": {"content-type": "application/json", "user-agent": "..."},
  "query": {"source": "checkout"},        # the query string, if any
  "received_at": "2026-09-16T10:31:07Z",
  "script": {"id": 4, "name": "Shop orders"}
}

def main(event):
    rows = db["shop"].sql("SELECT status, count(*) AS n FROM orders GROUP BY status")

    # Shown with the run, as JSON. Nothing is written from it.
    return {row["status"]: row["n"] for row in rows}

def main(event):
    # Only paid orders matter here. Anything else is a success that did
    # nothing — not an error, and nothing to retry.
    if event["event"] != "order.paid":
        return None

    db["shop"].sql("UPDATE orders SET status = ? WHERE code = ?",
           ["paid", event["payload"]["id"]])

def main(event):
    order = event["payload"]
    existing = db["shop"].get("orders", code=order["id"])

    if existing:
        db["shop"].sql("UPDATE orders SET status = ?, total = ? WHERE id = ?",
               [order["status"], order["total"], existing["id"]])
    else:
        db["shop"].sql("INSERT INTO orders (code, status, total) VALUES (?, ?, ?)",
               [order["id"], order["status"], order["total"]])

def main(event):
    order = event["payload"]

    # PostgreSQL. Needs a unique index on orders.code — add it on the schema page.
    db["shop"].sql(
        "INSERT INTO orders (code, status, total) VALUES (?, ?, ?) "
        "ON CONFLICT (code) DO UPDATE "
        "SET status = excluded.status, total = excluded.total",
        [order["id"], order["status"], order["total"]],
    )

def main(event):
    db["shop"].sql("INSERT INTO orders (code) VALUES (?)", ["A-1"])   # written, now
    raise ValueError("something went wrong down here")

# The row is in the table. A statement is written when it runs, not when
# main finishes — so write handlers that can safely run twice. Look the
# row up first, or use ON CONFLICT, and Run again cannot leave two.

def main(event):
    lines = event["payload"]["lines"]

    if not lines:
        return

    # One statement for the lot. Only the placeholders are joined into the
    # text — "(?, ?), (?, ?), ..." — and every value still goes in params.
    db["shop"].sql(
        "INSERT INTO order_lines (sku, quantity) VALUES "
        + ", ".join(["(?, ?)"] * len(lines)),
        [value for line in lines for value in (line["sku"], line["qty"])],
    )

def main(event):
    order = event["payload"]

    # One event, two tables: keep the customer current and record the order.
    if not db["shop"].get("customers", email=order["email"]):
        db["shop"].sql("INSERT INTO customers (email, name) VALUES (?, ?)",
               [order["email"], order["customer"]])

    if not db["shop"].get("orders", code=order["id"]):
        db["shop"].sql("INSERT INTO orders (code, email, total) VALUES (?, ?, ?)",
               [order["id"], order["email"], order["total"]])

# One address, three events. This is the whole reason a script takes a list
# of events rather than one: the sender has a single URL field to fill in.
def main(event):
    order = event["payload"]
    name = event["event"]

    if name == "order.paid":
        if db["shop"].get("orders", code=order["id"]):
            db["shop"].sql("UPDATE orders SET status = ?, total = ? WHERE code = ?",
                   ["paid", order["total"], order["id"]])
        else:
            db["shop"].sql("INSERT INTO orders (code, status, total) VALUES (?, ?, ?)",
                   [order["id"], "paid", order["total"]])

    elif name == "order.shipped":
        db["shop"].sql("UPDATE orders SET status = ?, tracking = ? WHERE code = ?",
               ["shipped", order["tracking"], order["id"]])

    elif name == "order.refunded":
        db["shop"].sql("UPDATE orders SET status = ? WHERE code = ?",
               ["refunded", order["id"]])

    else:
        # An event this handler does not know about. Say so in the log and stop.
        print("ignoring", name)

# Stripe — a payment succeeded.
#
# Stripe names its events in the body under "type", which is one of the
# fallbacks this application checks, so nothing needs configuring: set the
# script's events to `payment_intent.succeeded` and point Stripe at the
# address as it is.
#
# Stripe sends an event again when it is not sure the first one landed, so
# the payment is looked up by its id before it is added.
def main(event):
    intent = event["payload"]["data"]["object"]

    if db["shop"].get("payments", intent_id=intent["id"]):
        db["shop"].sql("UPDATE payments SET status = ? WHERE intent_id = ?",
               [intent["status"], intent["id"]])
        return

    db["shop"].sql(
        "INSERT INTO payments (intent_id, amount, currency, email, status) "
        "VALUES (?, ?, ?, ?, ?)",
        [
            intent["id"],
            intent["amount"] / 100.0,
            intent["currency"].upper(),
            intent.get("receipt_email") or "",
            intent["status"],
        ],
    )

# GitHub — a push, and an issue opened.
#
# GitHub names the event in the `x-github-event` header, which this
# application reads, so `push` and `issues` arrive as they were sent. Putting
# the name at the end of the address works too, and the path always wins:
#
#     https://10lines.io/hooks/7/whin_.../push
def main(event):
    body = event["payload"]

    if event["event"] == "push":
        commits = body["commits"]

        if commits:
            # One statement for every commit in the push. ON CONFLICT needs a
            # unique index on commits.sha: a commit pushed to two branches
            # arrives twice and is kept once.
            db["shop"].sql(
                "INSERT INTO commits (sha, message, author, repo) VALUES "
                + ", ".join(["(?, ?, ?, ?)"] * len(commits))
                + " ON CONFLICT (sha) DO NOTHING",
                [
                    value
                    for commit in commits
                    for value in (
                        commit["id"],
                        commit["message"].split("\n")[0][:200],
                        commit["author"]["name"],
                        body["repository"]["full_name"],
                    )
                ],
            )

    elif event["event"] == "issues" and body["action"] == "opened":
        issue = body["issue"]

        if not db["shop"].get("issues", number=issue["number"]):
            db["shop"].sql(
                "INSERT INTO issues (number, title, author, url) VALUES (?, ?, ?, ?)",
                [issue["number"], issue["title"], issue["user"]["login"],
                 issue["html_url"]],
            )

# Shopify — an order was created or updated.
#
# Shopify sends the topic in the `x-shopify-topic` header, which this
# application reads, so the events arrive as `orders/create` and
# `orders/updated`. Looking the order up by Shopify's id is what makes the
# update arrive as an update.
def main(event):
    order = event["payload"]
    customer = order.get("customer") or {}

    values = [
        order.get("order_number"),
        order.get("email") or "",
        (customer.get("first_name", "") + " " + customer.get("last_name", "")).strip(),
        float(order.get("total_price") or 0),
        order.get("currency") or "",
        str(order["id"]),
    ]

    if db["shop"].get("orders", shopify_id=str(order["id"])):
        db["shop"].sql(
            "UPDATE orders SET number = ?, email = ?, customer = ?, total = ?, "
            "currency = ? WHERE shopify_id = ?",
            values,
        )
    else:
        db["shop"].sql(
            "INSERT INTO orders (number, email, customer, total, currency, shopify_id) "
            "VALUES (?, ?, ?, ?, ?, ?)",
            values,
        )

# A plain HTML form on a static site.
#
#   <form method="post" action="https://10lines.io/hooks/7/whin_.../contact">
#     <input name="name"><input name="email"><textarea name="message"></textarea>
#   </form>
#
# A form post arrives as ordinary fields, so there is nothing to parse. Trim
# what people type, refuse what is empty, and write the rest.
def main(event):
    form = event["payload"]
    message = (form.get("message") or "").strip()

    if not message:
        return None

    db["shop"].sql(
        "INSERT INTO messages (name, email, message, user_agent) VALUES (?, ?, ?, ?)",
        [
            (form.get("name") or "anonymous").strip()[:80],
            (form.get("email") or "").strip()[:200],
            message[:5000],
            event["headers"].get("user-agent", "")[:200],
        ],
    )

# A device posting a reading every minute.
#
# Two things worth doing when the sender is a device: refuse nonsense before it
# becomes a row, and keep the value in one unit whatever the device sends.
def main(event):
    reading = event["payload"]
    celsius = reading.get("c")

    if celsius is None and "f" in reading:
        celsius = (reading["f"] - 32) * 5 / 9

    if celsius is None or not (-90 <= celsius <= 60):
        print("refusing an impossible reading:", reading)
        return None

    db["shop"].sql(
        "INSERT INTO readings (device, celsius, taken_at) VALUES (?, ?, ?)",
        [
            event["query"].get("device", "unknown"),
            round(celsius, 2),
            reading.get("at") or event["received_at"],
        ],
    )

# One project feeding another.
#
# In the sending project, a script posts with `http(..., sign=...)`. That
# signs the body with a shared secret and names the event in `x-10lines-event`,
# which an address reads without being told to:
#
#     def main(event):
#         rows = db["shop"].sql("SELECT id, title FROM posts WHERE status = ?", ["published"])
#         http("https://10lines.io/hooks/7/whin_.../posts.sync",
#              json={"rows": rows}, sign="SYNC_SECRET")
#
# In this project, put the same secret in Secrets as SYNC_SECRET and choose
# "A shared secret" under the address. This is the receiving end:
def main(event):
    rows = event["payload"]["rows"]

    if not rows:
        return

    # One statement for the lot. ON CONFLICT needs a unique index on
    # mirrored_posts.source_id, and makes a second sync an update.
    db["shop"].sql(
        "INSERT INTO mirrored_posts (source_id, title) VALUES "
        + ", ".join(["(?, ?)"] * len(rows))
        + " ON CONFLICT (source_id) DO UPDATE SET title = excluded.title",
        [value for row in rows for value in (row["id"], row["title"])],
    )

main raised KeyError

Traceback (most recent call last):
  File "<handler>", line 2, in main
    db["shop"].sql("INSERT INTO posts (title) VALUES (?)", [event["payload"]["name"]])
                                                    ~~~~~~~~~~~~~~~~^^^^^^^^
KeyError: 'name'

# The fix is almost always the same one: the sender did not send what you
# expected. Reach for .get() with a default rather than [].
def main(event):
    payload = event["payload"]
    title = payload.get("name") or payload.get("title") or "untitled"

    db["shop"].sql("INSERT INTO posts (title) VALUES (?)", [title])