temporal_client.py

# temporal_version/client.py
import asyncio
import logging
from temporalio.client import Client
from temporalio.exceptions import WorkflowFailureError

from .workflows import OrderProcessingWorkflow, ProcessOrderRequest

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def execute_order_workflow(order_data: dict):
    """
    Execute the order processing workflow.
    
    This demonstrates:
    - Workflow execution from client
    - Error handling
    - Result retrieval
    - Workflow monitoring
    """
    
    # Connect to Temporal server
    client = await Client.connect("localhost:7233")
    
    try:
        # Create workflow request
        request = ProcessOrderRequest(
            order_id=order_data["order_id"],
            customer_id=order_data["customer_id"],
            items=order_data["items"],
            total_amount=order_data["total_amount"]
        )
        
        logger.info(f"Starting workflow for order {request.order_id}")
        
        # Start the workflow
        # Temporal will automatically assign a unique workflow ID
        handle = await client.start_workflow(
            OrderProcessingWorkflow.run,
            request,
            id=f"order-{request.order_id}",  # Unique workflow ID
            task_queue="order-processing-queue",
        )
        
        logger.info(f"Workflow started with ID: {handle.id}")
        logger.info(f"Workflow URL: http://localhost:8080/namespaces/default/workflows/{handle.id}")
        
        # Wait for workflow completion
        result = await handle.result()
        
        logger.info(f"Workflow completed!")
        logger.info(f"Final status: {result.status}")
        
        if result.failure_reason:
            logger.error(f"Failure reason: {result.failure_reason}")
        
        return result
        
    except WorkflowFailureError as e:
        logger.error(f"Workflow failed: {e}")
        return None
    finally:
        await client.close()

async def demonstrate_workflow_features():
    """
    Demonstrate key Temporal workflow features.
    """
    
    print("=== Temporal Order Processing Demo ===")
    print()
    
    # Sample orders for demonstration
    sample_orders = [
        {
            "order_id": "order_001",
            "customer_id": "customer_123",
            "items": [
                {"name": "laptop", "quantity": 1},
                {"name": "mouse", "quantity": 1}
            ],
            "total_amount": 1299.99
        },
        {
            "order_id": "order_002", 
            "customer_id": "customer_456",
            "items": [
                {"name": "keyboard", "quantity": 2},
                {"name": "monitor", "quantity": 1}
            ],
            "total_amount": 599.99
        }
    ]
    
    for order in sample_orders:
        print(f"Processing order: {order['order_id']}")
        print(f"Customer: {order['customer_id']}")
        print(f"Items: {order['items']}")
        print(f"Total: ${order['total_amount']}")
        print()
        
        result = await execute_order_workflow(order)
        
        if result:
            if result.status == "completed":
                print("✅ Order processed successfully!")
            else:
                print(f"❌ Order failed: {result.failure_reason}")
        else:
            print("❌ Workflow execution failed!")
        
        print("-" * 50)
        print()

async def demonstrate_failure_recovery():
    """
    Demonstrate Temporal's failure recovery capabilities.
    
    This shows how workflows can be interrupted and resumed.
    """
    
    print("=== Failure Recovery Demo ===")
    print("This demo shows how Temporal handles worker failures")
    print("Try stopping the worker (Ctrl+C) during execution and restarting it")
    print()
    
    # Create a long-running order
    long_order = {
        "order_id": "order_recovery_test",
        "customer_id": "customer_recovery",
        "items": [
            {"name": "laptop", "quantity": 3},
            {"name": "monitor", "quantity": 2}
        ],
        "total_amount": 4999.99
    }
    
    print("Starting long-running workflow...")
    print("Worker can be stopped and restarted - workflow will continue!")
    print()
    
    result = await execute_order_workflow(long_order)
    
    if result and result.status == "completed":
        print("✅ Workflow completed successfully despite any interruptions!")
    else:
        print("❌ Workflow failed")

async def query_workflow_status():
    """
    Demonstrate workflow queries for real-time status.
    """
    client = await Client.connect("localhost:7233")
    
    try:
        # Get handle to existing workflow
        workflow_id = "order-order_001"  # From previous execution
        
        try:
            handle = client.get_workflow_handle(workflow_id)
            
            # Query workflow status (if it implements queries)
            logger.info(f"Querying workflow {workflow_id}")
            
            # Note: This would work if we implemented queries in the workflow
            # status = await handle.query("get_status")
            # print(f"Current status: {status}")
            
        except Exception as e:
            logger.info(f"Workflow {workflow_id} not found or completed: {e}")
            
    finally:
        await client.close()

def print_temporal_benefits():
    """
    Print the key benefits of using Temporal for order processing.
    """
    
    print("\n" + "="*60)
    print("KEY TEMPORAL BENEFITS DEMONSTRATED")
    print("="*60)
    
    benefits = [
        "🔄 Automatic State Persistence: Workflow state is automatically saved",
        "🔁 Built-in Retries: Activities retry automatically with exponential backoff", 
        "⏱️ Timeout Management: Configurable timeouts for all operations",
        "🔍 Full Observability: Complete workflow history in Temporal Web UI",
        "🛡️ Failure Recovery: Workers can crash and restart without losing progress",
        "🏗️ Saga Pattern: Built-in compensation for complex transactions",
        "📊 Deterministic Execution: Workflows are replay-safe and predictable",
        "🚀 Horizontal Scaling: Multiple workers can process tasks concurrently",
        "🧪 Easy Testing: Deterministic workflows are easy to unit test",
        "📈 Production Ready: Built-in metrics, monitoring, and alerting"
    ]
    
    for benefit in benefits:
        print(f"  {benefit}")
    
    print("\n" + "="*60)
    print("VISIT: http://localhost:8080 to see workflow execution in real-time!")
    print("="*60)

async def main():
    """
    Main demo function showcasing different Temporal features.
    """
    
    print("Choose a demo:")
    print("1. Basic Order Processing")
    print("2. Failure Recovery Demo") 
    print("3. Query Workflow Status")
    print("4. Show Temporal Benefits")
    
    choice = input("\nEnter choice (1-4): ").strip()
    
    if choice == "1":
        await demonstrate_workflow_features()
    elif choice == "2":
        await demonstrate_failure_recovery()
    elif choice == "3":
        await query_workflow_status()
    elif choice == "4":
        print_temporal_benefits()
    else:
        print("Invalid choice. Running basic demo...")
        await demonstrate_workflow_features()
    
    print_temporal_benefits()

if __name__ == "__main__":
    asyncio.run(main())