Checkout example sources - Vexlo

Save the required files in the application directory containing the SDK’s .venv environment. The examples use fixed local customer, catalog, inventory and payment responses. They do not send payments or change inventory. Configure the application’s queue using the setup steps in function usage or the investigation before running the files.

Pricing example

Save order_service.py to record the checkout. Save test_pricing.py when testing its recorded input.

"""A checkout request using fixed local data instead of external dependencies."""

import json
import os

import vexlo

ORDERS = {
    "order-1042": {
        "order_id": "order-1042",
        "customer_id": "customer-17",
        "lines": [
            {"sku": "notebook", "quantity": 2},
            {"sku": "pen", "quantity": 3},
        ],
    }
}
CUSTOMERS = {"customer-17": {"customer_id": "customer-17", "loyalty_member": True}}
INVENTORY = {"notebook": 5, "pen": 12}
CATALOG_PRICES = {"notebook": 1200, "pen": 300}


class PaymentGatewayUnavailable(RuntimeError):
    pass


@vexlo.step
def validate_order(order):
    if not order["lines"]:
        raise ValueError("order has no items")
    for line in order["lines"]:
        if line["quantity"] < 1 or line["sku"] not in INVENTORY:
            raise ValueError("invalid order item")
    return {"valid": True, "line_count": len(order["lines"])}


@vexlo.step
def load_customer(customer_id):
    return CUSTOMERS[customer_id]


@vexlo.step
def unit_price(sku):
    return CATALOG_PRICES[sku]


@vexlo.step
def line_total(line):
    return line["quantity"] * unit_price(line["sku"])


@vexlo.step
def price_lines(lines):
    return sum(line_total(line) for line in lines)


@vexlo.step
def discount_for(customer, subtotal_cents):
    return subtotal_cents // 10 if customer["loyalty_member"] else 0


@vexlo.step
def tax_for(net_cents):
    return (net_cents * 8 + 50) // 100


@vexlo.step
def price_order(lines, customer):
    subtotal = price_lines(lines)
    discount = discount_for(customer, subtotal)
    tax = tax_for(subtotal - discount)
    return {
        "subtotal_cents": subtotal,
        "discount_cents": discount,
        "tax_cents": tax,
        "total_cents": subtotal - discount + tax,
    }


@vexlo.step
def reserve_inventory(lines):
    for line in lines:
        if INVENTORY[line["sku"]] < line["quantity"]:
            raise ValueError("insufficient inventory")
    # A fixed response models a reservation without changing inventory or calling a service.
    return {"reservation_id": "reservation-1042", "units": sum(line["quantity"] for line in lines)}


@vexlo.step(retries=True)
def authorize_payment(order_id, amount_cents, processor):
    # Fixed responses model primary failure and backup success without sending a payment.
    if processor == "primary":
        raise PaymentGatewayUnavailable("primary processor unavailable")
    return {"payment_id": "payment-1042", "amount_cents": amount_cents, "processor": processor}


@vexlo.step
def create_receipt(order_id, price, reservation, payment):
    return {
        "order_id": order_id,
        "total_cents": price["total_cents"],
        "payment_id": payment["payment_id"],
        "reservation_id": reservation["reservation_id"],
        "items_reserved": reservation["units"],
        "accepted": True,
    }


@vexlo.trace
def checkout(order_id, app_revision):
    order = ORDERS[order_id]
    validate_order(order)
    customer = load_customer(order["customer_id"])
    price = price_order(order["lines"], customer)
    reservation = reserve_inventory(order["lines"])
    try:
        payment = authorize_payment(order_id, price["total_cents"], "primary")
    except PaymentGatewayUnavailable:
        payment = authorize_payment(order_id, price["total_cents"], "backup")
    return create_receipt(order_id, price, reservation, payment)


def main():
    vexlo.init(
        service="checkout-graph-example",
        spool_dir=os.environ["VEXLO_SPOOL_DIR"],
        tag_keys=["order_id", "app_revision"],
    )
    print(json.dumps(checkout("order-1042", os.environ["APP_REVISION"]), sort_keys=True))
    vexlo.flush()


if __name__ == "__main__":
    main()

"""Build a pricing regression case from the selected trace and recorded values."""
import hashlib
import json
import os
from pathlib import Path
from urllib.parse import urlencode
from urllib.request import Request, urlopen

import order_service


def read_json(route):
    request = Request(
        os.environ["VEXLO_ENDPOINT"].rstrip("/") + route,
        headers={"Authorization": "Bearer " + os.environ["VEXLO_TOKEN"]},
    )
    with urlopen(request, timeout=10) as response:
        return json.load(response)


def decoded(values, path, slot):
    matches = [v for v in values if v["path"] == path and v["slot"] == slot]
    assert len(matches) == 1, "expected one attempt-specific value for this example"
    value = matches[0]
    assert value["state"] == "stored" and value["encoding"] == "json"
    assert value.get("truncated") is False and "value" in value
    return json.loads(value["value"])


trace = json.loads(Path("trace.json").read_text())
revision = "sha256:" + hashlib.sha256(Path(order_service.__file__).read_bytes()).hexdigest()
assert trace["tags"]["app_revision"] == revision, "source does not match the recording"
assert trace["next_cursor"] is None and len(trace["steps"]) == trace["step_count"]
price = next(s for s in trace["steps"] if s["name"] == "order_service.price_order")
lookups = [s for s in trace["steps"] if s["name"] == "order_service.unit_price"]
assert len(lookups) == 2
assert all(len(s["attempts"]) == 1 for s in [price, *lookups])
price_values = json.loads(Path("values.json").read_text())["values"]
inputs = decoded(price_values, price["path"], "input")
expected = decoded(price_values, price["path"], "output")
query = urlencode(
    [("path", s["path"]) for s in lookups] + [("slot", "input"), ("slot", "output")]
)
lookup_values = read_json("/v1/traces/" + trace["trace_id"] + "/values?" + query)["values"]
catalog_prices = {
    decoded(lookup_values, s["path"], "input")["sku"]:
        decoded(lookup_values, s["path"], "output")
    for s in lookups
}
order_service.CATALOG_PRICES = catalog_prices
assert order_service.price_order(**inputs) == expected
print("pricing input and observed catalog reproduce the recorded output")

Incorrect and corrected checkout

Save both files to follow the pricing investigation. Its instructions copy the selected version to checkout_service.py before running it.

"""A checkout request using fixed local data instead of external dependencies."""

import json
import os
import logging
import sys

import vexlo

ORDERS = {
    "order-1042": {
        "order_id": "order-1042",
        "customer_id": "customer-17",
        "lines": [
            {"sku": "notebook", "quantity": 2},
            {"sku": "pen", "quantity": 3},
        ],
    }
}
CUSTOMERS = {"customer-17": {"customer_id": "customer-17", "loyalty_member": True}}
INVENTORY = {"notebook": 5, "pen": 12}
CATALOG_PRICES = {"notebook": 1200, "pen": 300}


class PaymentGatewayUnavailable(RuntimeError):
    pass


@vexlo.step
def validate_order(order):
    if not order["lines"]:
        raise ValueError("order has no items")
    for line in order["lines"]:
        if line["quantity"] < 1 or line["sku"] not in INVENTORY:
            raise ValueError("invalid order item")
    return {"valid": True, "line_count": len(order["lines"])}


@vexlo.step
def load_customer(customer_id):
    return CUSTOMERS[customer_id]


@vexlo.step
def unit_price(sku):
    return CATALOG_PRICES[sku]


@vexlo.step
def line_total(line):
    return line["quantity"] * unit_price(line["sku"])


@vexlo.step
def price_lines(lines):
    return sum(line_total(line) for line in lines)


@vexlo.step
def discount_for(customer, subtotal_cents):
    return subtotal_cents // 10 if customer.get("is_member", False) else 0


@vexlo.step
def tax_for(net_cents):
    return (net_cents * 8 + 50) // 100


@vexlo.step
def price_order(lines, customer):
    subtotal = price_lines(lines)
    discount = discount_for(customer, subtotal)
    tax = tax_for(subtotal - discount)
    return {
        "subtotal_cents": subtotal,
        "discount_cents": discount,
        "tax_cents": tax,
        "total_cents": subtotal - discount + tax,
    }


@vexlo.step
def reserve_inventory(lines):
    for line in lines:
        if INVENTORY[line["sku"]] < line["quantity"]:
            raise ValueError("insufficient inventory")
    # A fixed response models a reservation without changing inventory or calling a service.
    return {"reservation_id": "reservation-1042", "units": sum(line["quantity"] for line in lines)}


@vexlo.step(retries=True)
def authorize_payment(order_id, amount_cents, processor):
    # Fixed responses model primary failure and backup success without sending a payment.
    if processor == "primary":
        raise PaymentGatewayUnavailable("primary processor unavailable")
    return {"payment_id": "payment-1042", "amount_cents": amount_cents, "processor": processor}


@vexlo.step
def create_receipt(order_id, price, reservation, payment):
    return {
        "order_id": order_id,
        "total_cents": price["total_cents"],
        "payment_id": payment["payment_id"],
        "reservation_id": reservation["reservation_id"],
        "items_reserved": reservation["units"],
        "accepted": True,
    }


@vexlo.trace
def checkout(order_id, app_revision):
    order = ORDERS[order_id]
    validate_order(order)
    customer = load_customer(order["customer_id"])
    price = price_order(order["lines"], customer)
    reservation = reserve_inventory(order["lines"])
    try:
        payment = authorize_payment(order_id, price["total_cents"], "primary")
    except PaymentGatewayUnavailable:
        payment = authorize_payment(order_id, price["total_cents"], "backup")
    return create_receipt(order_id, price, reservation, payment)


def main():
    vexlo.init(
        service="checkout-investigation",
        spool_dir=os.environ["VEXLO_SPOOL_DIR"],
        tag_keys=["order_id", "app_revision"],
    )
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
    receipt = checkout("order-1042", os.environ["APP_REVISION"])
    logging.info("order %s accepted=%s total_cents=%s", receipt["order_id"], receipt["accepted"], receipt["total_cents"])
    print(json.dumps(receipt, sort_keys=True))
    vexlo.flush()


if __name__ == "__main__":
    main()

"""A checkout request using fixed local data instead of external dependencies."""

import json
import os
import logging
import sys

import vexlo

ORDERS = {
    "order-1042": {
        "order_id": "order-1042",
        "customer_id": "customer-17",
        "lines": [
            {"sku": "notebook", "quantity": 2},
            {"sku": "pen", "quantity": 3},
        ],
    }
}
CUSTOMERS = {"customer-17": {"customer_id": "customer-17", "loyalty_member": True}}
INVENTORY = {"notebook": 5, "pen": 12}
CATALOG_PRICES = {"notebook": 1200, "pen": 300}


class PaymentGatewayUnavailable(RuntimeError):
    pass


@vexlo.step
def validate_order(order):
    if not order["lines"]:
        raise ValueError("order has no items")
    for line in order["lines"]:
        if line["quantity"] < 1 or line["sku"] not in INVENTORY:
            raise ValueError("invalid order item")
    return {"valid": True, "line_count": len(order["lines"])}


@vexlo.step
def load_customer(customer_id):
    return CUSTOMERS[customer_id]


@vexlo.step
def unit_price(sku):
    return CATALOG_PRICES[sku]


@vexlo.step
def line_total(line):
    return line["quantity"] * unit_price(line["sku"])


@vexlo.step
def price_lines(lines):
    return sum(line_total(line) for line in lines)


@vexlo.step
def discount_for(customer, subtotal_cents):
    return subtotal_cents // 10 if customer["loyalty_member"] else 0


@vexlo.step
def tax_for(net_cents):
    return (net_cents * 8 + 50) // 100


@vexlo.step
def price_order(lines, customer):
    subtotal = price_lines(lines)
    discount = discount_for(customer, subtotal)
    tax = tax_for(subtotal - discount)
    return {
        "subtotal_cents": subtotal,
        "discount_cents": discount,
        "tax_cents": tax,
        "total_cents": subtotal - discount + tax,
    }


@vexlo.step
def reserve_inventory(lines):
    for line in lines:
        if INVENTORY[line["sku"]] < line["quantity"]:
            raise ValueError("insufficient inventory")
    # A fixed response models a reservation without changing inventory or calling a service.
    return {"reservation_id": "reservation-1042", "units": sum(line["quantity"] for line in lines)}


@vexlo.step(retries=True)
def authorize_payment(order_id, amount_cents, processor):
    # Fixed responses model primary failure and backup success without sending a payment.
    if processor == "primary":
        raise PaymentGatewayUnavailable("primary processor unavailable")
    return {"payment_id": "payment-1042", "amount_cents": amount_cents, "processor": processor}


@vexlo.step
def create_receipt(order_id, price, reservation, payment):
    return {
        "order_id": order_id,
        "total_cents": price["total_cents"],
        "payment_id": payment["payment_id"],
        "reservation_id": reservation["reservation_id"],
        "items_reserved": reservation["units"],
        "accepted": True,
    }


@vexlo.trace
def checkout(order_id, app_revision):
    order = ORDERS[order_id]
    validate_order(order)
    customer = load_customer(order["customer_id"])
    price = price_order(order["lines"], customer)
    reservation = reserve_inventory(order["lines"])
    try:
        payment = authorize_payment(order_id, price["total_cents"], "primary")
    except PaymentGatewayUnavailable:
        payment = authorize_payment(order_id, price["total_cents"], "backup")
    return create_receipt(order_id, price, reservation, payment)


def main():
    vexlo.init(
        service="checkout-investigation",
        spool_dir=os.environ["VEXLO_SPOOL_DIR"],
        tag_keys=["order_id", "app_revision"],
    )
    logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")
    receipt = checkout("order-1042", os.environ["APP_REVISION"])
    logging.info("order %s accepted=%s total_cents=%s", receipt["order_id"], receipt["accepted"], receipt["total_cents"])
    print(json.dumps(receipt, sort_keys=True))
    vexlo.flush()


if __name__ == "__main__":
    main()