""" Example usage of Sales Order API endpoints. This script demonstrates how to interact with the sales order endpoints. Run the FastAPI server first: python main.py Then execute this script: python examples/sales_order_example.py """ import httpx import asyncio from datetime import datetime, timedelta from decimal import Decimal BASE_URL = "http://localhost:8000" HEADERS = { "Content-Type": "application/json", "X-Merchant-ID": "merchant_test_001", "X-User-ID": "user_test_001", "X-Correlation-ID": "test-correlation-123" } async def create_sample_order(): """Create a sample sales order.""" payload = { "order_number": f"SO-TEST-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}", "branch_id": "branch_456", "order_date": datetime.utcnow().isoformat(), "customer": { "customer_id": "cust_123", "customer_name": "ABC retail", "customer_type": "b2b", "phone": "+91-9876543210", "email": "contact@abcretail.com", "gstin": "29ABCDE1234F1Z5", "billing_address": { "line1": "123 Main Street", "city": "Bangalore", "state": "Karnataka", "postal_code": "560001", "country": "India" } }, "items": [ { "sku": "PRD-001", "product_id": "prod_789", "product_name": "Professional Shampoo 500ml", "item_type": "product", "quantity": 10, "unit_price": 499.00, "tax_percent": 18.00, "discount_percent": 5.00, "hsn_code": "33051000", "uom": "bottle" }, { "sku": "SRV-001", "product_id": "srv_456", "product_name": "Hair Styling Service", "item_type": "service", "quantity": 2, "unit_price": 2500.00, "tax_percent": 18.00, "discount_percent": 0.00, "hsn_code": "999631", "uom": "service", "staff_id": "staff_123", "staff_name": "John Doe" } ], "payment": { "payment_type": "credit", "credit_terms": "net_30" }, "shipment": { "expected_delivery_date": (datetime.utcnow() + timedelta(days=7)).isoformat(), "shipping_method": "standard" }, "notes": "Urgent order - please expedite", "status": "confirmed", "created_by": "user_test_001" } async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/sales/order", json=payload, headers=HEADERS, timeout=30.0 ) if response.status_code == 201: print("✓ Sales order created successfully!") result = response.json() print(f" Order ID: {result['data']['sales_order_id']}") print(f" Order Number: {result['data']['order_number']}") print(f" Grand Total: ₹{result['data']['summary']['grand_total']}") return result['data']['sales_order_id'] else: print(f"✗ Error creating order: {response.status_code}") print(f" {response.text}") return None async def get_order_details(sales_order_id: str): """Get sales order details.""" async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/sales/order/{sales_order_id}", headers=HEADERS, timeout=30.0 ) if response.status_code == 200: print("✓ Retrieved order details successfully!") order = response.json() print(f" Order Number: {order['order_number']}") print(f" Status: {order['status']}") print(f" Customer: {order['customer']['customer_name']}") print(f" Items: {len(order['items'])}") print(f" Grand Total: ₹{order['summary']['grand_total']}") return order else: print(f"✗ Error getting order: {response.status_code}") print(f" {response.text}") return None async def list_orders(): """List all sales orders.""" payload = { "filters": { "status": ["confirmed", "processing"] }, "pagination": { "page": 1, "limit": 10 }, "sort": { "field": "order_date", "order": "desc" } } async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/sales/order/list", json=payload, headers=HEADERS, timeout=30.0 ) if response.status_code == 200: print("✓ Retrieved order list successfully!") result = response.json() print(f" Total orders: {result['data']['total']}") print(f" Orders in this page: {len(result['data']['documents'])}") for doc in result['data']['documents']: print(f" - {doc['order_number']} | {doc['customer']['customer_name']} | ₹{doc['summary']['grand_total']}") return result['data'] else: print(f"✗ Error listing orders: {response.status_code}") print(f" {response.text}") return None async def generate_invoice_for_order(sales_order_id: str): """Generate invoice for a sales order.""" payload = { "invoice_date": datetime.utcnow().isoformat(), "due_date": (datetime.utcnow() + timedelta(days=30)).isoformat(), "notes": "Payment due in 30 days" } async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/sales/order/{sales_order_id}/invoice", json=payload, headers=HEADERS, timeout=30.0 ) if response.status_code == 200: print("✓ Invoice generated successfully!") result = response.json() print(f" Invoice ID: {result['data']['invoice_id']}") print(f" Invoice Number: {result['data']['invoice_number']}") print(f" Invoice PDF: {result['data']['invoice_pdf_url']}") return result['data'] else: print(f"✗ Error generating invoice: {response.status_code}") print(f" {response.text}") return None async def get_sales_widgets(): """Get sales dashboard widgets.""" async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/sales/info/widgets", headers=HEADERS, timeout=30.0 ) if response.status_code == 200: print("✓ Retrieved sales widgets successfully!") widgets = response.json() print(f" Total Sales: ₹{widgets['total_sales'][0]['value']}") print(f" Pending Orders: {widgets['pending_orders'][0]['value']}") print(f" Fulfilled Orders: {widgets['fulfilled_orders'][0]['value']}") print(f" Revenue: ₹{widgets['revenue'][0]['value']}") return widgets else: print(f"✗ Error getting widgets: {response.status_code}") print(f" {response.text}") return None async def main(): """Run all examples.""" print("\n" + "="*60) print("Sales Order API Example Usage") print("="*60 + "\n") # Check health print("1. Checking API health...") async with httpx.AsyncClient() as client: try: response = await client.get(f"{BASE_URL}/health", timeout=5.0) if response.status_code == 200: print("✓ API is healthy\n") else: print("✗ API health check failed") return except Exception as e: print(f"✗ Cannot connect to API: {e}") print(" Make sure the server is running: python main.py\n") return # Create order print("2. Creating a sample sales order...") sales_order_id = await create_sample_order() print() if not sales_order_id: return # Get order details print("3. Retrieving order details...") await get_order_details(sales_order_id) print() # List orders print("4. Listing all orders...") await list_orders() print() # Generate invoice print("5. Generating invoice...") await generate_invoice_for_order(sales_order_id) print() # Get widgets print("6. Fetching sales widgets...") await get_sales_widgets() print() print("="*60) print("All examples completed!") print("="*60 + "\n") if __name__ == "__main__": asyncio.run(main())