Error Handling & Troubleshooting

Copy

• Copy for LLM

  Copy page as Markdown for LLMs
• View as Markdown

  Open this page as Markdown
• Open in ChatGPT

  Get insights from ChatGPT
• Open in Claude

  Get insights from Claude
• Connect to Cursor

  Install MCP server on Cursor
• Connect to VS Code

  Install MCP server on VS Code

def handle_auth_error(response, token_manager):
    """Handle authentication errors with automatic recovery"""
    if response.status_code == 401:
        logger.warning("Authentication failed, refreshing token...")
        token_manager.force_refresh()
        return True  # Indicates retry needed
    return False

def resilient_api_call(url, headers, data=None):
    """API call with automatic auth recovery"""
    max_retries = 2
    
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data)
            
            if response.status_code == 401 and attempt < max_retries - 1:
                if handle_auth_error(response, token_manager):
                    headers['Authorization'] = f'Bearer {token_manager.get_valid_token()}'
                    continue  # Retry with new token
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed on attempt {attempt + 1}: {e}")
            if attempt == max_retries - 1:
                raise

def validate_headers(headers):
    """Validate required headers before making requests"""
    required_headers = ['User-Agent', 'x-api-key', 'Authorization', 'vbasoftware-database']
    
    for header in required_headers:
        if header not in headers:
            raise ValueError(f"Missing required header: {header}")
    
    # Ensure User-Agent is descriptive
    user_agent = headers.get('User-Agent', '')
    if not user_agent or user_agent == 'python-requests':
        raise ValueError("User-Agent must be descriptive (e.g., 'MyCompany-Integration/1.0')")

# Always validate before API calls
headers = get_authenticated_headers()
validate_headers(headers)
response = requests.post(url, headers=headers, json=data)

import time
import random

def exponential_backoff(attempt, max_attempts=5, base_delay=1, max_delay=60):
    """Implements exponential backoff with jitter"""
    if attempt >= max_attempts:
        raise Exception(f"Max retry attempts ({max_attempts}) reached")
    
    # Calculate delay: base_delay * (2 ^ attempt) + jitter
    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
    delay = min(delay, max_delay)  # Cap at max_delay
    
    logger.info(f"Rate limited, waiting {delay:.2f} seconds before retry {attempt + 1}")
    time.sleep(delay)

def robust_api_call(url, headers, data=None, max_retries=5):
    """API call with intelligent retry logic"""
    for attempt in range(max_retries):
        try:
            response = requests.post(url, headers=headers, json=data, timeout=30)
            
            # Handle rate limiting
            if response.status_code == 429:
                logger.warning(f"Rate limited, attempt {attempt + 1}/{max_retries}")
                exponential_backoff(attempt)
                continue
            
            # Handle auth errors
            if response.status_code == 401:
                if handle_auth_error(response, token_manager):
                    headers['Authorization'] = f'Bearer {token_manager.get_valid_token()}'
                    continue
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.Timeout:
            logger.warning(f"Request timeout on attempt {attempt + 1}")
            if attempt < max_retries - 1:
                exponential_backoff(attempt)
                continue
            raise
            
        except requests.exceptions.ConnectionError:
            logger.warning(f"Connection error on attempt {attempt + 1}")
            if attempt < max_retries - 1:
                exponential_backoff(attempt)
                continue
            raise
    
    raise Exception("All retry attempts failed")

def parse_validation_errors(error_response):
    """Extract validation details from 400 responses"""
    if error_response.get('error', {}).get('status') == 400:
        detail = error_response['error'].get('detail', '')
        print(f"Validation error: {detail}")
        
        # Log the activity ID for support
        activity_id = error_response.get('debug', {}).get('activityId')
        print(f"Activity ID: {activity_id}")
        
        return activity_id
    return None

def handle_server_error(response):
    """Handle server errors appropriately"""
    activity_id = response.get('debug', {}).get('activityId', 'N/A')
    
    if response.status_code in [500, 502, 503, 504]:
        print(f"Server error {response.status_code}")
        print(f"Activity ID for support: {activity_id}")
        
        # Log for monitoring
        logger.error(f"VBAPI server error: {response.status_code}, Activity: {activity_id}")
        
        # Don't retry 500s immediately - may indicate a persistent issue
        if response.status_code == 500:
            return False  # Don't retry
        
        # Retry 502, 503, 504 (likely transient)
        return True  # Retry after delay

import logging
import requests
from collections import defaultdict

logger = logging.getLogger(__name__)

class VBAPIErrorHandler:
    def __init__(self):
        self.error_counts = defaultdict(int)
    
    def handle_response(self, response, endpoint):
        """Centralized error handling with monitoring"""
        
        # Track error rates
        status = response.status_code
        self.error_counts[status] += 1
        
        # Extract activity ID
        activity_id = None
        try:
            if hasattr(response, 'json'):
                activity_id = response.json().get('debug', {}).get('activityId')
        except:
            pass
        
        # Handle specific error types
        if status == 401:
            return self._handle_auth_error(response, activity_id)
        elif status == 403:
            return self._handle_forbidden(response, activity_id)
        elif status == 429:
            return self._handle_rate_limit(response, activity_id)
        elif 400 <= status < 500:
            return self._handle_client_error(response, activity_id, endpoint)
        elif 500 <= status < 600:
            return self._handle_server_error(response, activity_id, endpoint)
        
        return "success"
    
    def _handle_auth_error(self, response, activity_id):
        logger.warning(f"Authentication failed: {activity_id}")
        return "refresh_token"
    
    def _handle_forbidden(self, response, activity_id):
        logger.error(f"Forbidden access: {activity_id}")
        return "check_permissions"
    
    def _handle_rate_limit(self, response, activity_id):
        logger.warning(f"Rate limited: {activity_id}")
        return "retry_with_backoff"
    
    def _handle_client_error(self, response, activity_id, endpoint):
        logger.error(f"Client error {response.status_code} on {endpoint}: {activity_id}")
        return "client_error"
    
    def _handle_server_error(self, response, activity_id, endpoint):
        logger.error(f"Server error {response.status_code} on {endpoint}: {activity_id}")
        
        # Alert on server errors
        if response.status_code == 500:
            self._alert_support(activity_id, endpoint, response.status_code)
        
        return "retry_if_transient"
    
    def _alert_support(self, activity_id, endpoint, status_code):
        """Alert mechanism for serious errors"""
        alert_message = f"VBAPI Error: {status_code} on {endpoint}, Activity: {activity_id}"
        # Send to monitoring system, email, Slack, etc.
        print(f"ALERT: {alert_message}")
    
    def get_error_stats(self):
        """Get current error statistics"""
        return dict(self.error_counts)

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60, expected_exception=Exception):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.expected_exception = expected_exception
        self.failure_count = 0
        self.last_failure_time = None
        self.state = CircuitState.CLOSED
    
    def call(self, func, *args, **kwargs):
        """Execute function with circuit breaker protection"""
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.timeout:
                self.state = CircuitState.HALF_OPEN
                self.failure_count = 0
            else:
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        """Handle successful call"""
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def _on_failure(self):
        """Handle failed call"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

# Usage
api_circuit_breaker = CircuitBreaker(failure_threshold=3, timeout=30)

def protected_api_call(url, headers, data):
    """API call protected by circuit breaker"""
    return api_circuit_breaker.call(
        requests.post, 
        url, 
        headers=headers, 
        json=data,
        timeout=30
    )

import uuid
from contextvars import ContextVar

# Thread-local storage for request tracking
request_context = ContextVar('request_context', default={})

class RequestTracker:
    def __init__(self):
        self.request_id = str(uuid.uuid4())
        self.activity_ids = []
    
    def add_activity_id(self, activity_id):
        """Track activity ID from VBAPI response"""
        if activity_id:
            self.activity_ids.append(activity_id)
    
    def get_correlation_data(self):
        """Get all correlation data for debugging"""
        return {
            'request_id': self.request_id,
            'activity_ids': self.activity_ids
        }

def track_api_call(func):
    """Decorator to automatically track API calls"""
    def wrapper(*args, **kwargs):
        # Get or create request tracker
        ctx = request_context.get()
        if 'tracker' not in ctx:
            ctx['tracker'] = RequestTracker()
            request_context.set(ctx)
        
        try:
            response = func(*args, **kwargs)
            
            # Extract and track activity ID
            if hasattr(response, 'json'):
                try:
                    data = response.json()
                    activity_id = data.get('debug', {}).get('activityId')
                    ctx['tracker'].add_activity_id(activity_id)
                except:
                    pass
            
            return response
        except Exception as e:
            # Log correlation data on errors
            correlation = ctx['tracker'].get_correlation_data()
            logger.error(f"API call failed: {e}, Correlation: {correlation}")
            raise
    
    return wrapper

# Usage
@track_api_call
def make_api_request(url, headers, data):
    return requests.post(url, headers=headers, json=data)

import time
import threading
from collections import deque, defaultdict

class VBAPIMonitor:
    def __init__(self, alert_threshold=0.1):  # 10% error rate
        self.alert_threshold = alert_threshold
        self.requests = deque()
        self.errors = deque()
        self.lock = threading.Lock()
        self.start_monitoring()
    
    def record_request(self, success=True, error_type=None):
        """Record API request result"""
        now = time.time()
        
        with self.lock:
            self.requests.append(now)
            if not success:
                self.errors.append((now, error_type))
            
            # Clean old data (last hour)
            cutoff = now - 3600
            while self.requests and self.requests[0] < cutoff:
                self.requests.popleft()
            while self.errors and self.errors[0][0] < cutoff:
                self.errors.popleft()
    
    def get_error_rate(self):
        """Calculate current error rate"""
        with self.lock:
            if not self.requests:
                return 0
            return len(self.errors) / len(self.requests)
    
    def check_alerts(self):
        """Check if error rate exceeds threshold"""
        error_rate = self.get_error_rate()
        request_count = len(self.requests)
        
        if error_rate > self.alert_threshold and request_count >= 10:
            self._send_alert(error_rate, request_count)
    
    def _send_alert(self, error_rate, request_count):
        """Send alert for high error rate"""
        message = f"VBAPI Error Rate Alert: {error_rate:.1%} ({len(self.errors)}/{request_count} requests failed)"
        logger.critical(message)
        # Send to monitoring system, Slack, email, etc.
    
    def start_monitoring(self):
        """Start background monitoring thread"""
        def monitor_loop():
            while True:
                time.sleep(60)  # Check every minute
                self.check_alerts()
        
        thread = threading.Thread(target=monitor_loop, daemon=True)
        thread.start()

# Global monitor instance
api_monitor = VBAPIMonitor()

def monitored_api_call(url, headers, data=None):
    """API call with automatic monitoring"""
    try:
        response = requests.post(url, headers=headers, json=data)
        response.raise_for_status()
        api_monitor.record_request(success=True)
        return response.json()
    except requests.exceptions.HTTPError as e:
        error_type = f"HTTP_{e.response.status_code}"
        api_monitor.record_request(success=False, error_type=error_type)
        raise
    except Exception as e:
        error_type = type(e).__name__
        api_monitor.record_request(success=False, error_type=error_type)
        raise