Exception Handling

class PapermillException(Exception):
    """
    Base exception for all papermill-related errors.
    
    Raised when an exception is encountered when operating on a notebook.
    """

class PapermillExecutionError(PapermillException):
    """
    Raised when an exception is encountered in a notebook during execution.
    
    Attributes:
    - cell_index: Index of the cell where error occurred
    - exec_count: Execution count of the failed cell
    - source: Source code of the failed cell
    - ename: Exception name
    - evalue: Exception value/message
    - traceback: Full traceback of the exception
    """
    
    def __init__(
        self,
        cell_index: int,
        exec_count: int,
        source: str,
        ename: str,
        evalue: str,
        traceback: list[str]
    ): ...

class PapermillMissingParameterException(PapermillException):
    """
    Raised when a parameter without a value is required to operate on a notebook.
    """

class PapermillRateLimitException(PapermillException):
    """
    Raised when an I/O request has been rate limited.
    """

class PapermillOptionalDependencyException(PapermillException):
    """
    Raised when an exception is encountered when an optional plugin is missing.
    """

class PapermillWarning(Warning):
    """
    Base warning for papermill operations.
    """

class PapermillParameterOverwriteWarning(PapermillWarning):
    """
    Warning when callee overwrites caller argument to pass down the stream.
    """

class AwsError(Exception):
    """
    Raised when an AWS Exception is encountered.
    """

class FileExistsError(AwsError):
    """
    Raised when a File already exists on S3.
    """

def missing_dependency_generator(package: str, dep: str) -> callable:
    """
    Generates a function that raises PapermillOptionalDependencyException
    for missing dependencies.
    
    Parameters:
    - package: Name of the package that requires the dependency
    - dep: Name of the dependency that is missing
    
    Returns:
    callable: Function that raises the appropriate exception when called
    """

def missing_environment_variable_generator(package: str, env_key: str) -> callable:
    """
    Generates a function that raises PapermillOptionalDependencyException
    for missing environment variables.
    
    Parameters:
    - package: Name of the package that requires the environment variable
    - env_key: Name of the missing environment variable
    
    Returns:
    callable: Function that raises the appropriate exception when called
    """

import papermill as pm
from papermill import PapermillExecutionError, PapermillMissingParameterException

try:
    pm.execute_notebook(
        'analysis.ipynb',
        'output.ipynb',
        parameters={'threshold': 0.8}
    )
except PapermillExecutionError as e:
    print(f"Execution failed in cell {e.cell_index}")
    print(f"Error: {e.ename}: {e.evalue}")
    print(f"Cell source:\n{e.source}")
    
except PapermillMissingParameterException as e:
    print(f"Missing required parameter: {e}")
    
except Exception as e:
    print(f"Unexpected error: {e}")

import papermill as pm
from papermill import PapermillExecutionError

def analyze_execution_error(notebook_path, output_path, parameters):
    try:
        return pm.execute_notebook(notebook_path, output_path, parameters=parameters)
    
    except PapermillExecutionError as e:
        error_info = {
            'notebook': notebook_path,
            'failed_cell_index': e.cell_index,
            'execution_count': e.exec_count,
            'error_type': e.ename,
            'error_message': e.evalue,
            'cell_source': e.source,
            'full_traceback': e.traceback
        }
        
        # Log detailed error information
        print(f"EXECUTION FAILED: {notebook_path}")
        print(f"Cell {e.cell_index} (execution count {e.exec_count})")
        print(f"Error: {e.ename}: {e.evalue}")
        print("=" * 75)
        print("Cell Source:")
        print(e.source)
        print("=" * 75)
        print("Traceback:")
        for line in e.traceback:
            print(line)
        
        return error_info

# Usage
result = analyze_execution_error(
    'problematic_notebook.ipynb',
    'output.ipynb',
    {'param1': 'value1'}
)

import papermill as pm
from papermill import PapermillMissingParameterException

class ParameterValidationError(Exception):
    """Custom exception for parameter validation errors."""
    pass

def validate_and_execute(notebook_path, output_path, parameters):
    # Get expected parameters
    expected_params = pm.inspect_notebook(notebook_path)
    
    # Check for required parameters
    required_params = [
        name for name, info in expected_params.items()
        if info['default'] is None
    ]
    
    missing_params = set(required_params) - set(parameters.keys())
    if missing_params:
        raise PapermillMissingParameterException(
            f"Missing required parameters: {missing_params}"
        )
    
    # Validate parameter types (example)
    for param_name, param_value in parameters.items():
        if param_name in expected_params:
            expected_type = expected_params[param_name]['inferred_type_name']
            if expected_type == 'int' and not isinstance(param_value, int):
                raise ParameterValidationError(
                    f"Parameter '{param_name}' should be int, got {type(param_value)}"
                )
    
    try:
        return pm.execute_notebook(notebook_path, output_path, parameters=parameters)
    except Exception as e:
        print(f"Execution failed: {e}")
        raise

# Usage
try:
    result = validate_and_execute(
        'notebook.ipynb',
        'output.ipynb',
        {'required_param': 'value', 'numeric_param': 42}
    )
except (PapermillMissingParameterException, ParameterValidationError) as e:
    print(f"Parameter error: {e}")

import papermill as pm
from papermill import PapermillException, AwsError

def robust_notebook_execution(input_path, output_path, parameters):
    max_retries = 3
    retry_count = 0
    
    while retry_count < max_retries:
        try:
            return pm.execute_notebook(input_path, output_path, parameters=parameters)
            
        except AwsError as e:
            print(f"AWS error (attempt {retry_count + 1}): {e}")
            if "rate limit" in str(e).lower():
                import time
                time.sleep(2 ** retry_count)  # Exponential backoff
                retry_count += 1
                continue
            else:
                raise
                
        except FileNotFoundError as e:
            print(f"File not found: {e}")
            raise
            
        except PapermillException as e:
            print(f"Papermill error: {e}")
            raise
            
        except Exception as e:
            print(f"Unexpected error: {e}")
            if retry_count < max_retries - 1:
                retry_count += 1
                continue
            else:
                raise
    
    raise Exception(f"Failed after {max_retries} attempts")

# Usage
try:
    result = robust_notebook_execution(
        's3://bucket/input.ipynb',
        's3://bucket/output.ipynb',
        {'param': 'value'}
    )
except Exception as e:
    print(f"Final failure: {e}")

import papermill as pm
from papermill import PapermillOptionalDependencyException

def execute_with_optional_features(notebook_path, output_path, use_s3=False):
    try:
        if use_s3:
            # This might raise PapermillOptionalDependencyException if boto3 is missing
            return pm.execute_notebook(
                f's3://bucket/{notebook_path}',
                f's3://bucket/{output_path}'
            )
        else:
            return pm.execute_notebook(notebook_path, output_path)
            
    except PapermillOptionalDependencyException as e:
        print(f"Optional dependency missing: {e}")
        print("Falling back to local execution...")
        
        # Fallback to local execution
        return pm.execute_notebook(notebook_path, output_path)

# Usage
result = execute_with_optional_features(
    'notebook.ipynb',
    'output.ipynb',
    use_s3=True
)

import logging
import papermill as pm
from papermill import PapermillExecutionError, PapermillException

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def monitored_execution(notebook_path, output_path, parameters):
    try:
        logger.info(f"Starting execution: {notebook_path}")
        result = pm.execute_notebook(notebook_path, output_path, parameters=parameters)
        logger.info(f"Execution completed successfully: {output_path}")
        return result
        
    except PapermillExecutionError as e:
        logger.error(f"Execution failed in cell {e.cell_index}: {e.ename}: {e.evalue}")
        logger.error(f"Failed cell source: {e.source}")
        
        # Send alert/notification
        send_execution_alert(notebook_path, e)
        raise
        
    except Exception as e:
        logger.error(f"Unexpected error executing {notebook_path}: {str(e)}")
        raise

def send_execution_alert(notebook_path, error):
    # Implement your alerting logic here
    alert_data = {
        'notebook': notebook_path,
        'error_type': error.ename,
        'error_message': error.evalue,
        'cell_index': error.cell_index,
        'timestamp': str(datetime.now())
    }
    # Send to monitoring system, email, Slack, etc.
    pass

# Usage
try:
    result = monitored_execution(
        'critical_analysis.ipynb',
        'results.ipynb',
        {'dataset': 'production_data.csv'}
    )
except Exception:
    # Error already logged and alert sent
    pass