Error Handling

{
  "error": {
    "code": "FILE_001",
    "message": "Invalid file type",
    "details": {
      "supported_types": ["PDF", "JPG", "PNG", "DOCX", "PPTX", "XLSX", "HTML"],
      "received_type": "DOC"
    }
  }
}

class PulseAPIError(Exception):
    """Custom exception for Pulse API errors."""
    
    def __init__(self, code, message, details=None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(f"{code}: {message}")

class ErrorHandler:
    """Centralized error handling for Pulse API."""
    
    # Retryable error codes
    RETRYABLE_CODES = {
        "FILE_005",     # Download failed
        "FILE_006",     # Download timeout
        "PROC_001",     # Processing failed
        "PROC_002",     # Processing timeout
        "PROC_003",     # Service unavailable
        "PROC_004",     # Rate limit
        "JOB_003",      # Job expired
        "STORAGE_003",  # Storage unavailable
        "GENERAL_001",  # Server error
        "GENERAL_002",  # Service unavailable
        "GENERAL_003",  # Gateway timeout
        "GENERAL_004",  # Rate limit
    }
    
    @staticmethod
    def handle_response(response):
        """Process API response and raise appropriate errors."""
        
        if response.status_code == 200:
            return response.json()
        
        # Parse error response
        try:
            error_data = response.json().get("error", {})
            code = error_data.get("code", str(response.status_code))
            message = error_data.get("message", "Unknown error")
            details = error_data.get("details", {})
        except:
            code = str(response.status_code)
            message = response.text or "Unknown error"
            details = {}
        
        # Determine if retryable
        is_retryable = code in ErrorHandler.RETRYABLE_CODES
        
        # Create appropriate exception
        error = PulseAPIError(code, message, details)
        error.is_retryable = is_retryable
        
        raise error

from datetime import datetime, timedelta

class CircuitBreaker:
    """Prevent cascading failures with circuit breaker."""
    
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "closed"  # closed, open, half-open
    
    def call(self, func):
        """Execute function with circuit breaker protection."""
        
        if self.state == "open":
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.recovery_timeout):
                self.state = "half-open"
                self.failure_count = 0
            else:
                raise Exception("Circuit breaker is open")
        
        try:
            result = func()
            if self.state == "half-open":
                self.state = "closed"
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = datetime.now()
            
            if self.failure_count >= self.failure_threshold:
                self.state = "open"
                print(f"Circuit breaker opened after {self.failure_count} failures")
            
            raise

# Usage
breaker = CircuitBreaker()
try:
    result = breaker.call(lambda: client.extract(file_url="document.pdf"))
except Exception as e:
    print(f"Failed: {e}")

class SmartRetry:
    """Intelligent retry with different strategies per error type."""
    
    def __init__(self):
        self.strategies = {
            "FILE_005": self.retry_with_backoff,       # Download failed
            "FILE_006": self.retry_with_backoff,       # Download timeout
            "PROC_002": self.retry_with_smaller_chunk, # Processing timeout
            "PROC_004": self.handle_rate_limit,        # Rate limit exceeded
            "JOB_003": self.retry_with_smaller_chunk,  # Job expired
            "GENERAL_004": self.handle_rate_limit,     # Too many requests
            "GENERAL_002": self.retry_with_backoff,    # Service unavailable
        }
    
    def execute(self, func, context=None):
        """Execute with smart retry logic."""
        
        max_attempts = 3
        
        for attempt in range(max_attempts):
            try:
                return func()
            except PulseAPIError as e:
                if attempt == max_attempts - 1:
                    raise
                
                strategy = self.strategies.get(e.code, self.retry_with_backoff)
                strategy(e, attempt, context)
    
    def retry_with_backoff(self, error, attempt, context):
        """Standard exponential backoff."""
        delay = 2 ** attempt
        print(f"Retrying after {delay}s due to {error.code}")
        time.sleep(delay)
    
    def handle_rate_limit(self, error, attempt, context):
        """Handle rate limiting with longer delay."""
        print("Rate limited. Waiting 60 seconds...")
        time.sleep(60)
    
    def retry_with_smaller_chunk(self, error, attempt, context):
        """Retry with smaller page range for timeouts."""
        if context and 'pages' in context:
            # Reduce page range
            current_pages = context['pages']
            # Logic to split page range
            print(f"Retrying with smaller page range")
        time.sleep(5)

def extract_with_fallback(file_path, preferred_mode="full"):
    """Extract with graceful degradation."""
    
    strategies = [
        # Try full extraction with schema
        lambda: client.extract(
            file_url=file_path,
            schema=complex_schema
        ),
        # Fallback to simple extraction
        lambda: client.extract(
            file_url=file_path,
            schema=simple_schema
        ),
        # Last resort: text only
        lambda: client.extract(
            file_url=file_path
        )
    ]
    
    for i, strategy in enumerate(strategies):
        try:
            print(f"Attempting strategy {i + 1}/{len(strategies)}")
            return strategy()
        except PulseAPIError as e:
            if i == len(strategies) - 1:
                raise
            print(f"Strategy {i + 1} failed: {e.code}, trying next...")

def process_large_document_with_recovery(file_path, total_pages=100):
    """Process document in chunks with partial success."""
    
    chunk_size = 10
    results = []
    failed_chunks = []
    
    for start in range(0, total_pages, chunk_size):
        end = min(start + chunk_size - 1, total_pages - 1)
        page_range = f"{start + 1}-{end + 1}"
        
        try:
            result = client.extract(
                file_url=file_path,
                pages=page_range
            )
            results.append({
                "pages": page_range,
                "content": result
            })
        except PulseAPIError as e:
            print(f"Failed to process pages {page_range}: {e}")
            failed_chunks.append(page_range)
    
    # Retry failed chunks with different strategy
    for chunk in failed_chunks:
        try:
            # Try with smaller chunks or different parameters
            result = client.extract(
                file_url=file_path,
                pages=chunk
            )
            results.append({
                "pages": chunk,
                "content": result,
                "recovered": True
            })
        except:
            print(f"Permanently failed: {chunk}")
    
    return results

def upload_with_validation(file_path):
    """Upload file with pre-validation."""
    
    # Check file extension
    valid_extensions = ['.pdf', '.jpg', '.jpeg', '.png', '.docx', '.pptx', '.xlsx', '.html']
    file_ext = os.path.splitext(file_path)[1].lower()
    
    if file_ext not in valid_extensions:
        raise ValueError(f"Unsupported file type: {file_ext}")
    
    # Attempt upload with retry
    return exponential_backoff_retry(
        lambda: client.extract(file_url=file_url)
    )

def manage_async_job(job_id):
    """Robustly manage async job lifecycle."""
    
    max_poll_time = 600  # 10 minutes
    poll_interval = 5
    start_time = time.time()
    
    while time.time() - start_time < max_poll_time:
        try:
            status = client.jobs.get_job(job_id=job_id)
            
            if status['status'] == 'completed':
                return status['result']
            elif status['status'] == 'failed':
                raise PulseAPIError(
                    "JOB_004",
                    f"Job failed: {status.get('error', 'Unknown error')}"
                )
            elif status['status'] == 'cancelled':
                raise PulseAPIError("JOB_002", "Job was cancelled")
            
            time.sleep(poll_interval)
            
        except PulseAPIError as e:
            if e.code == "JOB_001":
                # Job not found - might be eventual consistency issue
                time.sleep(10)
                continue
            raise
        except requests.exceptions.RequestException:
            # Network error - retry
            time.sleep(poll_interval)
            continue
    
    # Timeout - attempt to cancel
    try:
        client.jobs.cancel_job(job_id=job_id)
    except:
        pass
    
    raise TimeoutError(f"Job {job_id} timed out after {max_poll_time}s")