> ## Documentation Index
> Fetch the complete documentation index at: https://docs.vexlo.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Checkout example sources

> Self-contained Python sources for the execution graph, function-usage test and pricing investigation.

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](https://docs.vexlo.ai/guides/function-usage.md#record-the-example-optional) or the
[investigation](https://docs.vexlo.ai/guides/investigation.md#record-the-example-optional) before running the files.

## Pricing example

Save `order_service.py` to
[record the checkout](https://docs.vexlo.ai/guides/function-usage.md#record-the-example-optional).
Save `test_pricing.py` when
[testing its recorded input](https://docs.vexlo.ai/guides/function-usage.md#retrieve-an-input-and-use-it-in-a-test).

```python order_service.py theme={"system"}
"""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()
```

```python test_pricing.py theme={"system"}
"""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](https://docs.vexlo.ai/guides/investigation.md#record-the-example-optional).
Its instructions copy the selected version to `checkout_service.py` before running it.

```python checkout_service.before.py theme={"system"}
"""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()
```

```python checkout_service.after.py theme={"system"}
"""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()
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.