simple_happy_path_client.py

#   Copyright 2026 UCP Authors
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

"""Simple Happy Path Client using UCPClient class.

This script demonstrates a basic "happy path" user journey using the UCPClient:
0. Discovery: Querying the merchant to see what they support.
1. Creating a new checkout session (cart).
2. Adding items to the checkout session.
3. Applying a discount code.
4. Triggering fulfillment option generation.
5. Selecting a fulfillment destination.
6. Selecting a fulfillment option.
7. Completing the checkout by processing a payment.

Usage:
  uv run simple_happy_path_client.py --server_url=http://localhost:8182
"""

import argparse
import asyncio
import logging

from ucp_client import UCPClient
from ucp_sdk.models.schemas.shopping import checkout_create_req
from ucp_sdk.models.schemas.shopping import checkout_update_req
from ucp_sdk.models.schemas.shopping import payment_create_req
from ucp_sdk.models.schemas.shopping.types import buyer
from ucp_sdk.models.schemas.shopping.types import item_create_req
from ucp_sdk.models.schemas.shopping.types import item_update_req
from ucp_sdk.models.schemas.shopping.types import line_item_create_req
from ucp_sdk.models.schemas.shopping.types import line_item_update_req
from ucp_sdk.models.schemas.shopping.types.postal_address import PostalAddress


def get_payment_dict(checkout_obj):
    """Helper to get payment as dict from checkout (handles both dict and model)."""
    if not checkout_obj.payment:
        return {}
    if isinstance(checkout_obj.payment, dict):
        return checkout_obj.payment
    return checkout_obj.payment.model_dump(mode="json", by_alias=True, exclude_none=True)


async def run_happy_path(server_url: str) -> None:
    """Run the happy path checkout flow using UCPClient."""
    
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
    )
    logger = logging.getLogger(__name__)

    async with UCPClient(base_url=server_url) as client:
        # ======================================================================
        # STEP 0: Discovery
        # ======================================================================
        logger.info("STEP 0: Discovery - Asking merchant what they support...")
        
        profile = await client.get_merchant_profile()
        supported_handlers = profile.payment.handlers if profile.payment else []
        
        logger.info("Merchant supports %d payment handlers:", len(supported_handlers))
        for h in supported_handlers:
            logger.info(" - %s (%s)", h.id, h.name)

        # ======================================================================
        # STEP 1: Create a Checkout Session
        # ======================================================================
        logger.info("\nSTEP 1: Creating a new Checkout Session...")

        # We start with one item: "Red Rose"
        item1 = item_create_req.ItemCreateRequest(
            id="bouquet_roses", title="Red Rose"
        )
        line_item1 = line_item_create_req.LineItemCreateRequest(
            quantity=1, item=item1
        )

        # Convert handlers to dict format for payment request
        handlers_dict = [{"id": h.id, "name": h.name} for h in supported_handlers]
        
        payment_req = payment_create_req.PaymentCreateRequest(
            instruments=[],
            selected_instrument_id=None,
            handlers=handlers_dict,
        )

        buyer_req = buyer.Buyer(full_name="John Doe", email="john.doe@example.com")

        create_request = checkout_create_req.CheckoutCreateRequest(
            currency="USD",
            line_items=[line_item1],
            payment=payment_req,
            buyer=buyer_req,
        )

        checkout = await client.create_checkout(create_request)
        checkout_id = checkout.id
        
        logger.info("Successfully created checkout session: %s", checkout_id)
        logger.info("Current Total: %s cents", checkout.totals[-1].amount if checkout.totals else "N/A")

        # ======================================================================
        # STEP 2: Add More Items (Update Checkout)
        # ======================================================================
        logger.info("\nSTEP 2: Adding a second item (Ceramic Pot)...")

        # Update Item 1 (Roses) - Keep quantity 1
        item1_update = item_update_req.ItemUpdateRequest(
            id="bouquet_roses", title="Red Rose"
        )
        line_item1_update = line_item_update_req.LineItemUpdateRequest(
            id=checkout.line_items[0].id,
            quantity=1,
            item=item1_update,
        )

        # Add Item 2 (Ceramic Pot) - Quantity 2
        item2_update = item_update_req.ItemUpdateRequest(
            id="pot_ceramic", title="Ceramic Pot"
        )
        line_item2_update = line_item_update_req.LineItemUpdateRequest(
            quantity=2,
            item=item2_update,
        )

        # Get payment dict from current checkout
        payment_dict = get_payment_dict(checkout)

        update_request = checkout_update_req.CheckoutUpdateRequest(
            id=checkout_id,
            line_items=[line_item1_update, line_item2_update],
            currency=checkout.currency,
            payment=payment_dict,
        )

        checkout = await client.update_checkout(checkout_id, update_request)
        
        logger.info("Successfully added items.")
        logger.info("New Total: %s cents", checkout.totals[-1].amount if checkout.totals else "N/A")
        logger.info("Item Count: %d", len(checkout.line_items))

        # ======================================================================
        # STEP 3: Apply Discount
        # ======================================================================
        logger.info("\nSTEP 3: Applying Discount (10%% OFF)...")

        # Re-construct line items
        li_1 = next(
            li for li in checkout.line_items if li.item.id == "bouquet_roses"
        )
        li_2 = next(
            li for li in checkout.line_items if li.item.id == "pot_ceramic"
        )

        item1_update = item_update_req.ItemUpdateRequest(
            id="bouquet_roses", title="Red Rose"
        )
        line_item1_update = line_item_update_req.LineItemUpdateRequest(
            id=li_1.id,
            quantity=1,
            item=item1_update,
        )

        item2_update = item_update_req.ItemUpdateRequest(
            id="pot_ceramic", title="Ceramic Pot"
        )
        line_item2_update = line_item_update_req.LineItemUpdateRequest(
            id=li_2.id,
            quantity=2,
            item=item2_update,
        )

        payment_dict = get_payment_dict(checkout)

        update_request = checkout_update_req.CheckoutUpdateRequest(
            id=checkout_id,
            line_items=[line_item1_update, line_item2_update],
            currency=checkout.currency,
            payment=payment_dict,
        )

        update_dict = update_request.model_dump(mode="json", by_alias=True, exclude_none=True)
        update_dict["discounts"] = {"codes": ["10OFF"]}

        # Use raw _make_request for this since we need to add extra fields
        checkout_data = await client._make_request(
            "PUT", f"/checkout-sessions/{checkout_id}", update_dict
        )
        
        from ucp_sdk.models.schemas.shopping import checkout_resp
        checkout = checkout_resp.CheckoutResponse(**checkout_data)

        logger.info("Successfully applied discount.")
        logger.info("New Total: %s cents", checkout.totals[-1].amount if checkout.totals else "N/A")

        discounts_applied = checkout.discounts.applied if checkout.discounts and hasattr(checkout.discounts, 'applied') else []
        if discounts_applied:
            logger.info("Applied Discounts: %s", [d.code for d in discounts_applied])
        else:
            logger.warning("No discounts applied!")

        # ======================================================================
        # STEP 4: Trigger Fulfillment Option Generation
        # ======================================================================
        logger.info("\nSTEP 4: Selecting Fulfillment Option...")

        # Helper to get fulfillment methods (handles both dict and model)
        def get_fulfillment_methods(checkout_obj):
            if not checkout_obj.fulfillment:
                return None
            if isinstance(checkout_obj.fulfillment, dict):
                return checkout_obj.fulfillment.get("methods")
            return checkout_obj.fulfillment.methods if hasattr(checkout_obj.fulfillment, 'methods') else None

        # Check if fulfillment options need to be generated
        has_fulfillment = get_fulfillment_methods(checkout)
        
        if not has_fulfillment:
            logger.info("STEP 4: Triggering fulfillment option generation...")

            li_1 = next(
                li for li in checkout.line_items if li.item.id == "bouquet_roses"
            )
            li_2 = next(
                li for li in checkout.line_items if li.item.id == "pot_ceramic"
            )

            item1_update = item_update_req.ItemUpdateRequest(
                id="bouquet_roses", title="Red Rose"
            )
            line_item1_update = line_item_update_req.LineItemUpdateRequest(
                id=li_1.id,
                quantity=1,
                item=item1_update,
            )

            item2_update = item_update_req.ItemUpdateRequest(
                id="pot_ceramic", title="Ceramic Pot"
            )
            line_item2_update = line_item_update_req.LineItemUpdateRequest(
                id=li_2.id,
                quantity=2,
                item=item2_update,
            )

            payment_dict = get_payment_dict(checkout)

            trigger_req = checkout_update_req.CheckoutUpdateRequest(
                id=checkout_id,
                line_items=[line_item1_update, line_item2_update],
                currency=checkout.currency,
                payment=payment_dict,
                fulfillment={"methods": [{"type": "shipping"}]},
            )

            checkout = await client.update_checkout(checkout_id, trigger_req)

        # ======================================================================
        # STEP 5: Select Destination
        # ======================================================================
        fulfillment_methods = get_fulfillment_methods(checkout)
        if fulfillment_methods:
            method = fulfillment_methods[0]
            destinations = method.get("destinations") if isinstance(method, dict) else (method.destinations if hasattr(method, 'destinations') else None)
            
            if destinations:
                dest_id = destinations[0].get("id") if isinstance(destinations[0], dict) else destinations[0].id
                logger.info("STEP 5: Selecting destination: %s", dest_id)

                li_1 = next(
                    li for li in checkout.line_items if li.item.id == "bouquet_roses"
                )
                li_2 = next(
                    li for li in checkout.line_items if li.item.id == "pot_ceramic"
                )

                item1_update = item_update_req.ItemUpdateRequest(
                    id="bouquet_roses", title="Red Rose"
                )
                line_item1_update = line_item_update_req.LineItemUpdateRequest(
                    id=li_1.id,
                    quantity=1,
                    item=item1_update,
                )

                item2_update = item_update_req.ItemUpdateRequest(
                    id="pot_ceramic", title="Ceramic Pot"
                )
                line_item2_update = line_item_update_req.LineItemUpdateRequest(
                    id=li_2.id,
                    quantity=2,
                    item=item2_update,
                )

                payment_dict = get_payment_dict(checkout)

                dest_req = checkout_update_req.CheckoutUpdateRequest(
                    id=checkout_id,
                    line_items=[line_item1_update, line_item2_update],
                    currency=checkout.currency,
                    payment=payment_dict,
                    fulfillment={
                        "methods": [
                            {"type": "shipping", "selected_destination_id": dest_id}
                        ]
                    },
                )

                checkout = await client.update_checkout(checkout_id, dest_req)

                # ======================================================================
                # STEP 6: Select Option
                # ======================================================================
                fulfillment_methods = get_fulfillment_methods(checkout)
                method = fulfillment_methods[0]
                groups = method.get("groups") if isinstance(method, dict) else (method.groups if hasattr(method, 'groups') else None)
                if groups and groups[0]:
                    options = groups[0].get("options") if isinstance(groups[0], dict) else (groups[0].options if hasattr(groups[0], 'options') else None)
                    if options:
                        option_id = options[0].get("id") if isinstance(options[0], dict) else options[0].id
                        logger.info("STEP 6: Selecting option: %s", option_id)

                        li_1 = next(
                            li for li in checkout.line_items if li.item.id == "bouquet_roses"
                        )
                        li_2 = next(
                            li for li in checkout.line_items if li.item.id == "pot_ceramic"
                        )

                        item1_update = item_update_req.ItemUpdateRequest(
                            id="bouquet_roses", title="Red Rose"
                        )
                        line_item1_update = line_item_update_req.LineItemUpdateRequest(
                            id=li_1.id,
                            quantity=1,
                            item=item1_update,
                        )

                        item2_update = item_update_req.ItemUpdateRequest(
                            id="pot_ceramic", title="Ceramic Pot"
                        )
                        line_item2_update = line_item_update_req.LineItemUpdateRequest(
                            id=li_2.id,
                            quantity=2,
                            item=item2_update,
                        )

                        payment_dict = get_payment_dict(checkout)

                        option_req = checkout_update_req.CheckoutUpdateRequest(
                            id=checkout_id,
                            line_items=[line_item1_update, line_item2_update],
                            currency=checkout.currency,
                            payment=payment_dict,
                            fulfillment={
                                "methods": [{
                                    "type": "shipping",
                                    "selected_destination_id": dest_id,
                                    "groups": [{"selected_option_id": option_id}],
                                }]
                            },
                        )

                        checkout = await client.update_checkout(checkout_id, option_req)

                        logger.info("Fulfillment option selected.")
                        logger.info("Updated Total: %s cents", checkout.totals[-1].amount if checkout.totals else "N/A")

        # ======================================================================
        # STEP 7: Complete Checkout (Payment)
        # ======================================================================
        logger.info("\nSTEP 7: Processing Payment...")

        target_handler = "mock_payment_handler"
        handler_ids = [h.id for h in supported_handlers]
        
        if target_handler not in handler_ids:
            logger.error("Merchant does not support %s. Aborting.", target_handler)
            return

        billing_address = PostalAddress(
            street_address="123 Main St",
            address_locality="Anytown",
            address_region="CA",
            address_country="US",
            postal_code="12345",
        )

        payment_data = {
            "id": "instr_my_card",
            "handler_id": target_handler,
            "handler_name": target_handler,
            "type": "card",
            "brand": "Visa",
            "last_digits": "4242",
            "credential": {
                "type": "token",
                "token": "success_token"
            },
            "billing_address": billing_address.model_dump(mode="json", by_alias=True, exclude_none=True),
        }

        risk_signals = {
            "ip": "127.0.0.1",
            "browser": "python-httpx",
        }

        final_checkout = await client.complete_checkout(
            checkout_id=checkout_id,
            payment_data=payment_data,
            risk_signals=risk_signals,
        )

        logger.info("Payment Successful!")
        logger.info("Checkout Status: %s", final_checkout.status)
        logger.info("Order ID: %s", final_checkout.order.id if final_checkout.order else "N/A")
        logger.info("Order Permalink: %s", final_checkout.order.permalink_url if final_checkout.order else "N/A")

        # ======================================================================
        # DONE
        # ======================================================================
        logger.info("\nHappy Path completed successfully.")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--server_url",
        default="http://localhost:8182",
        help="Base URL of the UCP Server",
    )
    args = parser.parse_args()

    asyncio.run(run_happy_path(args.server_url))


if __name__ == "__main__":
    main()