""" ═══════════════════════════════════════════════════════════ 🐍 OPHION - Example Python Application This app is automatically instrumented by OpenTelemetry No code changes needed - just env vars! ═══════════════════════════════════════════════════════════ """ import os import time import random import logging from flask import Flask, jsonify # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) @app.route('/') def home(): """Home endpoint""" logger.info("Home endpoint called") return jsonify({ 'message': 'Hello from Python!', 'instrumented': True, 'otel_endpoint': os.getenv('OTEL_EXPORTER_OTLP_ENDPOINT', 'not set') }) @app.route('/health') def health(): """Health check endpoint""" return jsonify({ 'status': 'healthy', 'service': 'python-example' }) @app.route('/api/users') def get_users(): """Get users - simulates DB call""" logger.info("Fetching users") # Simulate some work time.sleep(random.uniform(0.01, 0.1)) return jsonify([ {'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}, {'id': 3, 'name': 'Charlie'} ]) @app.route('/api/slow') def slow_endpoint(): """Slow endpoint - good for testing trace visualization""" logger.info("Starting slow operation") # Simulate slow operation time.sleep(0.5) logger.info("Slow operation completed") return jsonify({ 'message': 'Slow response', 'delay': '500ms' }) @app.route('/api/chain') def chain_endpoint(): """Simulates a chain of operations""" logger.info("Starting chain operation") # Step 1: Validate time.sleep(random.uniform(0.01, 0.05)) logger.info("Validation complete") # Step 2: Process time.sleep(random.uniform(0.05, 0.1)) logger.info("Processing complete") # Step 3: Save time.sleep(random.uniform(0.02, 0.08)) logger.info("Save complete") return jsonify({ 'message': 'Chain completed', 'steps': ['validate', 'process', 'save'] }) @app.route('/api/error') def error_endpoint(): """Endpoint that raises an error - good for testing error traces""" logger.error("About to raise an error") raise ValueError("Intentional error for testing traces") @app.errorhandler(Exception) def handle_error(error): """Global error handler""" logger.exception(f"Unhandled error: {error}") return jsonify({ 'error': str(error), 'type': type(error).__name__ }), 500 if __name__ == '__main__': port = int(os.getenv('PORT', 5000)) logger.info(f"🚀 Python server starting on port {port}") logger.info(f"📊 OTEL Endpoint: {os.getenv('OTEL_EXPORTER_OTLP_ENDPOINT', 'not configured')}") logger.info(f"🏷️ Service Name: {os.getenv('OTEL_SERVICE_NAME', 'unknown')}") app.run(host='0.0.0.0', port=port)