Order Validation & Processing¶

Order Validation¶

Before processing, each order is validated against inventory.

Contract¶

`validate_order`

Validate an order before processing. Checks inventory for each item and calculates the final price. Returns a validation result with availability and pricing details.

Inputs

Parameter Type Default

customer_id str —

items list —

Output

return ValidationResult —

ValidationResult Order validation result.

Parameter	Type	Default
`valid` *	`bool`	—
`reason`	`str \| None`	None
`unavailable_items`	`list[dict]`	[]
`pricing`	`OrderTotal \| None`	None

Source Code¶

def validate_order(customer_id: str, items: list) -> ValidationResult:
    """Validate an order before processing.

    Checks inventory for each item and calculates the final price.
    Returns a validation result with availability and pricing details.
    """
    unavailable = []
    for item in items:
        try:
            stock = CheckInventory().call(item["product_id"])
            if not stock.available or stock.stock < item["quantity"]:
                unavailable.append({
                    "product_id": item["product_id"],
                    "requested": item["quantity"],
                    "available": stock.stock,
                })
        except ServiceError:
            unavailable.append({
                "product_id": item["product_id"],
                "requested": item["quantity"],
                "available": 0,
                "error": "Could not verify stock",
            })

    if unavailable:
        return ValidationResult(
            valid=False,
            reason="Some items are unavailable",
            unavailable_items=unavailable,
        )

    pricing = calculate_order_total(customer_id, items)

    return ValidationResult(valid=True, pricing=pricing)

Execution Flow¶

Sequence diagram — validate_order

sequenceDiagram
    participant validate_order
    participant calculate_order_total
    validate_order->>+calculate_order_total: customer_id, items
    calculate_order_total-->>-validate_order: result

Try it Live¶

Playground — `validate_order`

Validate an order before processing. Checks inventory for each item and calculates the final price. Returns a validation result with availability and pricing details.

View source code

def validate_order(customer_id: str, items: list) -> ValidationResult:
    """Validate an order before processing.

    Checks inventory for each item and calculates the final price.
    Returns a validation result with availability and pricing details.
    """
    unavailable = []
    for item in items:
        try:
            stock = CheckInventory().call(item["product_id"])
            if not stock.available or stock.stock < item["quantity"]:
                unavailable.append({
                    "product_id": item["product_id"],
                    "requested": item["quantity"],
                    "available": stock.stock,
                })
        except ServiceError:
            unavailable.append({
                "product_id": item["product_id"],
                "requested": item["quantity"],
                "available": 0,
                "error": "Could not verify stock",
            })

    if unavailable:
        return ValidationResult(
            valid=False,
            reason="Some items are unavailable",
            unavailable_items=unavailable,
        )

    pricing = calculate_order_total(customer_id, items)

    return ValidationResult(valid=True, pricing=pricing)

customer_id (str) *

items (list) *

Mock configuration

Warehouse / CheckInventory — Check product availability in the warehouse (Blackbox).

available (bool)

stock (int)

warehouse (str)

Simulate error

error_code

error_message

http_code

CustomerService / FetchCustomerTier — Retrieve customer loyalty tier (Blackbox).

tier (str)

discount_pct (int)

member_since (str)

Simulate error

error_code

error_message

http_code

ProductCatalog / FetchProductCatalog — Retrieve product details from the catalog service (Blackbox).

id (str)

name (str)

base_price (float)

category (str)

Simulate error

error_code

error_message

http_code

Test cases

No test cases saved yet.

Full Order Processing¶

The complete order flow: validate, price, and charge.

Contract¶

`process_order`

Process a complete order: validate, price, and charge. Orchestrates inventory check, pricing, and payment in sequence.

Inputs

Parameter Type Default

customer_id str —

items list —

payment_method str —

Output

return OrderResult —

OrderResult Final order processing result.

Parameter	Type	Default
`status` *	`str`	—
`reason`	`str \| None`	None
`transaction_id`	`str \| None`	None
`total_charged`	`float \| None`	None
`pricing`	`OrderTotal \| None`	None
`details`	`list[dict]`	[]

Execution Flow¶

Sequence diagram — process_order

sequenceDiagram
    participant process_order
    participant validate_order
    process_order->>+validate_order: customer_id, items
    validate_order-->>-process_order: result

Source Code¶

def process_order(customer_id: str, items: list, payment_method: str) -> OrderResult:
    """Process a complete order: validate, price, and charge.

    Orchestrates inventory check, pricing, and payment in sequence.
    """
    validation = validate_order(customer_id, items)

    if not validation.valid:
        return OrderResult(
            status="REJECTED",
            reason=validation.reason,
            details=validation.unavailable_items,
        )

    total = validation.pricing.total

    try:
        payment = ProcessPayment().call({
            "amount": total,
            "method": payment_method,
            "customer_id": customer_id,
        })
    except ServiceError:
        return OrderResult(
            status="PAYMENT_FAILED",
            reason="Payment could not be processed",
        )

    return OrderResult(
        status="CONFIRMED",
        transaction_id=payment.transaction_id,
        total_charged=payment.amount,
        pricing=validation.pricing,
    )

Try it Live¶

Playground — `process_order`

Process a complete order: validate, price, and charge. Orchestrates inventory check, pricing, and payment in sequence.

View source code

def process_order(customer_id: str, items: list, payment_method: str) -> OrderResult:
    """Process a complete order: validate, price, and charge.

    Orchestrates inventory check, pricing, and payment in sequence.
    """
    validation = validate_order(customer_id, items)

    if not validation.valid:
        return OrderResult(
            status="REJECTED",
            reason=validation.reason,
            details=validation.unavailable_items,
        )

    total = validation.pricing.total

    try:
        payment = ProcessPayment().call({
            "amount": total,
            "method": payment_method,
            "customer_id": customer_id,
        })
    except ServiceError:
        return OrderResult(
            status="PAYMENT_FAILED",
            reason="Payment could not be processed",
        )

    return OrderResult(
        status="CONFIRMED",
        transaction_id=payment.transaction_id,
        total_charged=payment.amount,
        pricing=validation.pricing,
    )

customer_id (str) *

items (list) *

payment_method (str) *

Mock configuration

Warehouse / CheckInventory — Check product availability in the warehouse (Blackbox).

available (bool)

stock (int)

warehouse (str)

Simulate error

error_code

error_message

http_code

CustomerService / FetchCustomerTier — Retrieve customer loyalty tier (Blackbox).

tier (str)

discount_pct (int)

member_since (str)

Simulate error

error_code

error_message

http_code

ProductCatalog / FetchProductCatalog — Retrieve product details from the catalog service (Blackbox).

id (str)

name (str)

base_price (float)

category (str)

Simulate error

error_code

error_message

http_code

PaymentGateway / ProcessPayment — Submit payment to the payment gateway (Blackbox).

status (str)

transaction_id (str)

amount (float)

Simulate error

error_code

error_message

http_code

Test cases

No test cases saved yet.

Order Validation & Processing¶