""" DevServer API Client Demonstrates usage of all API endpoints with Redis and Celery integration """ import requests import time import json import logging from typing import Optional, Dict, Any from datetime import datetime # Setup client logging logging.basicConfig( level=logging.INFO, format='%(asctime)s | %(levelname)-8s | %(message)s', datefmt='%H:%M:%S' ) logger = logging.getLogger("devserver_client") class DevServerClient: """Client for interacting with DevServer API""" def __init__(self, base_url: str = "http://localhost:8000"): self.base_url = base_url self.session = requests.Session() logger.info(f"DevServer client initialized: {base_url}") def get_api_info(self) -> Dict[str, Any]: """Get API information and available endpoints""" logger.debug("Fetching API information...") response = self.session.get(f"{self.base_url}/") response.raise_for_status() logger.info("API information retrieved") return response.json() def roll_dice(self) -> Dict[str, Any]: """Roll dice and get results""" logger.debug("Rolling dice...") response = self.session.get(f"{self.base_url}/api/dice/roll") response.raise_for_status() result = response.json() total_rolls = result.get('total', 0) logger.info(f"Dice rolled: {total_rolls} total rolls") return result def create_fibonacci_task(self, n: int) -> Dict[str, Any]: """ Create a Celery task to calculate Fibonacci number Args: n: Number to calculate (0-40) Returns: Dict containing task_id and status """ logger.debug(f"Creating Fibonacci task: n={n}") response = self.session.post(f"{self.base_url}/api/tasks/fibonacci?n={n}") response.raise_for_status() result = response.json() logger.info(f"Fibonacci task created: task_id={result['task_id']}") return result def get_task_result(self, task_id: str) -> Dict[str, Any]: """ Get the result of a Celery task Args: task_id: Task ID returned from create_fibonacci_task Returns: Dict containing task status and result (if completed) """ logger.debug(f"Checking task status: task_id={task_id}") response = self.session.get(f"{self.base_url}/api/tasks/{task_id}") response.raise_for_status() result = response.json() status = result.get('status') logger.debug(f"Task status: {status}") return result def wait_for_task(self, task_id: str, timeout: int = 30, poll_interval: float = 1.0) -> Dict[str, Any]: """ Wait for a task to complete with timeout Args: task_id: Task ID to wait for timeout: Maximum seconds to wait poll_interval: Seconds between status checks Returns: Final task result Raises: TimeoutError: If task doesn't complete within timeout """ logger.info(f"Waiting for task completion: task_id={task_id}, timeout={timeout}s") start_time = time.time() checks = 0 while time.time() - start_time < timeout: checks += 1 result = self.get_task_result(task_id) if result.get("status") == "completed": elapsed = time.time() - start_time logger.info(f"Task completed after {elapsed:.2f}s ({checks} checks)") return result time.sleep(poll_interval) logger.error(f"Task timeout after {timeout}s") raise TimeoutError(f"Task {task_id} did not complete within {timeout} seconds") def set_cache(self, key: str, value: str, ttl: int = 300) -> Dict[str, Any]: """ Set a value in Redis cache Args: key: Cache key value: Value to store ttl: Time to live in seconds (default 300) Returns: Dict containing success message """ logger.debug(f"Setting cache: key={key}, ttl={ttl}s") response = self.session.post( f"{self.base_url}/api/cache/set", params={"key": key, "value": value, "ttl": ttl} ) response.raise_for_status() logger.info(f"Cache set: key={key}") return response.json() def get_cache(self, key: str) -> Optional[Dict[str, Any]]: """ Get a value from Redis cache Args: key: Cache key Returns: Dict containing key and value, or None if not found """ logger.debug(f"Getting cache: key={key}") try: response = self.session.get(f"{self.base_url}/api/cache/{key}") response.raise_for_status() logger.info(f"Cache hit: key={key}") return response.json() except requests.exceptions.HTTPError as e: if e.response.status_code == 404: logger.warning(f"Cache miss: key={key}") return None raise def process_data(self, data: Dict[str, Any]) -> Dict[str, Any]: """ Process data with Redis caching and Celery Args: data: Dictionary of data to process Returns: Dict with source (cache/celery_task) and result or task_id """ logger.debug(f"Processing data: {len(data)} keys") response = self.session.post( f"{self.base_url}/api/process", json=data ) response.raise_for_status() result = response.json() source = result.get('source') logger.info(f"Data processing initiated: source={source}") return result def demo_basic_endpoints(): """Demonstrate basic API endpoints""" print("=" * 60) print("DEMO: Basic Endpoints") print("=" * 60) client = DevServerClient() # 1. Get API info print("\n1. Getting API information...") info = client.get_api_info() print(f" API: {info['message']}") print(f" Available endpoints: {len(info['endpoints'])}") # 2. Roll dice print("\n2. Rolling dice...") result = client.roll_dice() print(f" Dice rolls: {result['dice_rolls']}") print(f" Total rolls: {sum(len(rolls) for rolls in result['dice_rolls'])}") def demo_celery_tasks(): """Demonstrate Celery task functionality""" print("\n" + "=" * 60) print("DEMO: Celery Tasks (Fibonacci)") print("=" * 60) client = DevServerClient() # Calculate multiple Fibonacci numbers fibonacci_numbers = [5, 10, 15, 20] print("\n1. Submitting Fibonacci tasks...") tasks = [] for n in fibonacci_numbers: result = client.create_fibonacci_task(n) tasks.append((n, result['task_id'])) print(f" Fibonacci({n}): Task ID = {result['task_id']}") print("\n2. Waiting for tasks to complete...") for n, task_id in tasks: try: result = client.wait_for_task(task_id, timeout=30) fib_result = result['result'] print(f" Fibonacci({n}) = {fib_result['fibonacci']}") except TimeoutError as e: print(f" ERROR: {e}") def demo_redis_cache(): """Demonstrate Redis caching functionality""" print("\n" + "=" * 60) print("DEMO: Redis Cache Operations") print("=" * 60) client = DevServerClient() # 1. Set cache values print("\n1. Setting cache values...") cache_data = [ ("user:1", "Alice", 60), ("user:2", "Bob", 60), ("config:theme", "dark", 300), ("session:abc123", "active", 3600), ] for key, value, ttl in cache_data: result = client.set_cache(key, value, ttl) print(f" Set {key} = {value} (TTL: {ttl}s)") # 2. Get cache values print("\n2. Getting cache values...") for key, expected_value, _ in cache_data: result = client.get_cache(key) if result: print(f" Get {key} = {result['value']}") else: print(f" Get {key} = NOT FOUND") # 3. Try to get non-existent key print("\n3. Getting non-existent key...") result = client.get_cache("nonexistent:key") print(f" Result: {result}") def demo_combined_workflow(): """Demonstrate combined Redis + Celery workflow""" print("\n" + "=" * 60) print("DEMO: Combined Redis + Celery Workflow") print("=" * 60) client = DevServerClient() test_data = { "name": "test_user", "email": "test@example.com", "age": 25, "interests": ["coding", "music", "sports"] } # 1. First request - should process with Celery print("\n1. First request (cache miss - will process with Celery)...") result1 = client.process_data(test_data) print(f" Source: {result1['source']}") if result1['source'] == 'celery_task': task_id = result1['task_id'] print(f" Task ID: {task_id}") print(" Waiting for task to complete...") task_result = client.wait_for_task(task_id, timeout=30) print(f" Result: {json.dumps(task_result['result'], indent=2)}") # 2. Second request with same data - should return from cache print("\n2. Second request (cache hit - should be instant)...") time.sleep(2) # Wait a bit for cache to be set result2 = client.process_data(test_data) print(f" Source: {result2['source']}") if result2['source'] == 'cache': print(f" Cached Result: {json.dumps(result2['result'], indent=2)}") print(" Cache working correctly!") def demo_performance_comparison(): """Compare performance with and without caching""" print("\n" + "=" * 60) print("DEMO: Performance Comparison (Cache vs No Cache)") print("=" * 60) client = DevServerClient() # Test data test_data = { "operation": "performance_test", "data_size": "medium", "timestamp": time.time() } # First request (no cache) print("\n1. First request (processing with Celery)...") start_time = time.time() result1 = client.process_data(test_data) first_duration = 0.0 if result1['source'] == 'celery_task': task_result = client.wait_for_task(result1['task_id']) first_duration = time.time() - start_time print(f" Duration: {first_duration:.2f} seconds") else: print(" Unexpected: got cached result on first try") return # Second request (with cache) print("\n2. Second request (from cache)...") start_time = time.time() result2 = client.process_data(test_data) second_duration = time.time() - start_time print(f" Duration: {second_duration:.2f} seconds") if first_duration > 0: speedup = first_duration / second_duration if second_duration > 0 else 0 print(f"\n Cache Speedup: {speedup:.1f}x faster!") def run_simple_demo(): """Run a simple demo of the main features""" print("\n" + "=" * 60) print("DevServer API - Simple Demo") print("=" * 60) print("\nTesting: FastAPI + Redis + Celery + OpenTelemetry") print("View traces at: http://localhost:16686") print("=" * 60) start_time = time.time() try: logger.info("Starting DevServer demo...") client = DevServerClient() # 1. Test basic endpoint print("\n1. Testing dice roll endpoint...") logger.info("TEST 1: Dice roll") result = client.roll_dice() print(f" Dice rolls: {result['dice_rolls']}") logger.info(f"Result: {sum(len(r) for r in result['dice_rolls'])} dice rolled") # 2. Test Redis cache print("\n2. Testing Redis cache...") logger.info("TEST 2: Redis caching") test_key = f"test_key_{int(time.time())}" test_value = f"test_value_{datetime.now().isoformat()}" client.set_cache(test_key, test_value, 60) cached = client.get_cache(test_key) if cached: print(f" Cached value: {cached['value']}") logger.info(f"Cache verified: key={test_key}") else: print(" Cache not found") logger.warning("Cache verification failed") # 3. Test Celery task print("\n3. Testing Celery task (Fibonacci)...") logger.info("TEST 3: Celery task") task = client.create_fibonacci_task(10) print(f" Task submitted: {task['task_id']}") logger.info(f"Fibonacci task submitted: task_id={task['task_id']}") result = client.wait_for_task(task['task_id'], timeout=30) fib_value = result['result']['fibonacci'] calc_time = result['result'].get('calculation_time', 'N/A') print(f" Fibonacci(10) = {fib_value} (calculated in {calc_time})") logger.info(f"Fibonacci result: {fib_value}") elapsed = time.time() - start_time print("\n" + "=" * 60) print(f"All tests passed in {elapsed:.2f}s!") print("Check Jaeger UI at: http://localhost:16686") print("=" * 60) logger.info(f"Demo completed successfully in {elapsed:.2f}s") except requests.exceptions.ConnectionError: print("\nERROR: Cannot connect to DevServer") print(" Run: docker-compose up -d") logger.error("Connection error: DevServer not reachable") except Exception as e: print(f"\nERROR: {e}") logger.error(f"Demo failed: {e}", exc_info=True) import traceback traceback.print_exc() # Uncomment to run all detailed demos # def run_all_demos(): # """Run all demonstration scenarios""" # demo_basic_endpoints() # time.sleep(1) # demo_redis_cache() # time.sleep(1) # demo_celery_tasks() # time.sleep(1) # demo_combined_workflow() # time.sleep(1) # demo_performance_comparison() if __name__ == "__main__": run_simple_demo() # Uncomment to run all detailed demos: # run_all_demos()