11_Handling_exceptions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Handling Exceptions\n",
    "\n",
    "In this notebook, we'll explore how to handle exceptions effectively. Exception handling is crucial for building robust and maintainable code, especially in complex workflows. We'll cover best practices, demonstrate how to implement them in a data science context, and illustrate advanced techniques such as using custom exceptions and ensuring clean error handling across nested functions.\n",
    "\n",
    "## Table of Contents\n",
    "\n",
    "1. [Basic Exception Handling](#1)\n",
    "2. [Custom Exceptions](#2)\n",
    "3. [Nested Functions and Exception Propagation](#3)\n",
    "4. [Logging Exceptions](#4)\n",
    "5. [Step-by-Step Example](#5)\n",
    "6. [Exercise](#6)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 1. Basic Exception Handling <a name=\"1\"></a>\n",
    "\n",
    "Exception handling allows your code to deal with errors gracefully. Here's a simple example of handling an exception in a data loading step."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "def load_data(filepath):\n",
    "    try:\n",
    "        data = pd.read_csv(filepath)\n",
    "        return data\n",
    "    except FileNotFoundError:\n",
    "        print(f\"Error: The file at {filepath} was not found.\")\n",
    "    except pd.errors.EmptyDataError:\n",
    "        print(\"Error: No data in file.\")\n",
    "    except Exception as e:\n",
    "        print(f\"An unexpected error occurred: {e}\")\n",
    "\n",
    "# Usage\n",
    "data = load_data('data/raw/non_existent_file.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 2. Custom Exceptions <a name=\"2\"></a>\n",
    "\n",
    "Creating custom exceptions allows you to handle specific error conditions more gracefully."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class DataPipelineError(Exception):\n",
    "    pass\n",
    "\n",
    "class DataValidationError(DataPipelineError):\n",
    "    def __init__(self):\n",
    "        super().__init__(\"Data validation failed.\")\n",
    "\n",
    "class MissingValuesError(DataPipelineError):\n",
    "    def __init__(self, missing_values):\n",
    "        self.missing_values = missing_values\n",
    "        super().__init__(\n",
    "            f\"Data contains {missing_values} missing values.\")\n",
    "\n",
    "# This function always raises an exception\n",
    "def validate_data(data):\n",
    "    missing_values = data.isnull().sum().sum()\n",
    "    if  missing_values > 0:\n",
    "        raise MissingValuesError(missing_values)\n",
    "    else:\n",
    "        raise DataValidationError\n",
    "\n",
    "# Usage\n",
    "try:\n",
    "    data = load_data('train.csv')\n",
    "    validate_data(data)\n",
    "except MissingValuesError as e:\n",
    "    print(f\"Data contains {e.missing_values} missing values.\")\n",
    "except DataValidationError as e:\n",
    "    print(f\"Validation Error: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 3. Nested Functions and Exception Propagation <a name=\"3\"></a>\n",
    "\n",
    "Handling exceptions in nested functions ensures that errors are caught and managed properly, preventing unexpected crashes.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class DataPreprocessingError(DataPipelineError):\n",
    "    def __init__(self):\n",
    "        super().__init__(\"Missing column during preprocessing\")\n",
    "\n",
    "def preprocess_data(data):\n",
    "    try:\n",
    "        # Example preprocessing step\n",
    "        data['new_column'] = data['existing_column'] * 2\n",
    "        return data\n",
    "    except KeyError as e:\n",
    "        raise DataPreprocessingError from e\n",
    "\n",
    "def run_pipeline(filepath):\n",
    "    try:\n",
    "        data = load_data(filepath)\n",
    "        validate_data(data)\n",
    "        data = preprocess_data(data)\n",
    "        return data\n",
    "    except DataPreprocessingError as e:\n",
    "        print(f\"Data validation failed: {e}\")\n",
    "        raise\n",
    "    except Exception as e:\n",
    "        print(f\"An unexpected error occurred in the pipeline: {e}\")\n",
    "        raise\n",
    "\n",
    "# Usage\n",
    "processed_data = run_pipeline('train.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 4. Logging Exceptions <a name=\"4\"></a>\n",
    "\n",
    "Using logging for exception handling provides a more flexible and powerful way to manage errors, especially in production environments."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import logging\n",
    "\n",
    "# Configure logging\n",
    "logging.basicConfig(level=logging.INFO)\n",
    "logger = logging.getLogger(__name__)\n",
    "\n",
    "def load_data(filepath):\n",
    "    try:\n",
    "        data = pd.read_csv(filepath)\n",
    "        return data\n",
    "    except FileNotFoundError:\n",
    "        logger.exception(f\"File not found: {filepath}\")\n",
    "        raise\n",
    "    except pd.errors.EmptyDataError:\n",
    "        logger.exception(\"No data in file.\")\n",
    "        raise\n",
    "    except Exception as e:\n",
    "        logger.exception(f\"An unexpected error occurred: {e}\")\n",
    "        raise\n",
    "\n",
    "# Usage\n",
    "try:\n",
    "    data = load_data('data/raw/non_existent_file.csv')\n",
    "except Exception as e:\n",
    "    logger.critical(f\"Critical error occurred: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 5. Step-by-Step Example <a name=\"5\"></a>\n",
    "\n",
    "We'll now build a complete data science pipeline with exception handling at each step.\n",
    "\n",
    "We start with this code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import mean_squared_error\n",
    "\n",
    "def load_data(filepath):\n",
    "    data = pd.read_csv(filepath)\n",
    "    return data\n",
    "    \n",
    "def preprocess_data(data):\n",
    "    X = data.drop(columns=['target'])\n",
    "    y = data['target']\n",
    "    return train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "\n",
    "def train_model(X_train, y_train):\n",
    "    model = LinearRegression()\n",
    "    model.fit(X_train, y_train)\n",
    "    return model\n",
    "\n",
    "def pipeline(datapath='train.csv'):\n",
    "    # Step 1: Load the data\n",
    "    data = load_data('data.csv')\n",
    "\n",
    "    # Step 2: Preprocess the data\n",
    "    X_train, X_test, y_train, y_test = preprocess_data(data)\n",
    "    \n",
    "    # Step 3: Train the model\n",
    "    model = train_model(X_train, y_train)\n",
    "\n",
    "    # Step 4: Evaluate the model\n",
    "    predictions = model.predict(X_test)\n",
    "    mse = mean_squared_error(y_test, predictions)\n",
    "    print(f\"Model Mean Squared Error: {mse}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Step 1: Setup up the logger and create a custom parent Exception**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import mean_squared_error\n",
    "import logging\n",
    "\n",
    "# Configure logging\n",
    "logging.basicConfig(level=logging.INFO)\n",
    "logger = logging.getLogger(__name__)\n",
    "\n",
    "# Create parent exception class\n",
    "class ModelPipelineError(Exception):\n",
    "    pass"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "**Step 2: Data Loading**\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "class FileMissingException(ModelPipelineError):\n",
    "    def __init__(self, filepath):\n",
    "        self.filepath = filepath\n",
    "        super().__init__(\n",
    "            f\"The file {self.filepath} couldn't be found.\")\n",
    "    \n",
    "def load_data(filepath):\n",
    "    try:\n",
    "        data = pd.read_csv(filepath)\n",
    "        return data\n",
    "    except FileNotFoundError as e:\n",
    "        raise FileMissingException(filepath) from e"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Step 3: Data Preprocessing**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TargetMissingException(ModelPipelineError):\n",
    "    def __init__(self):\n",
    "        super().__init__(\"The target is missing in the dataset.\")\n",
    "\n",
    "def preprocess_data(data):\n",
    "    try:\n",
    "        if 'target' not in data.columns:\n",
    "            raise TargetMissingException\n",
    "        data = data.dropna()  # Handle missing values\n",
    "        X = data.drop(columns=['target'])\n",
    "        y = data['target']\n",
    "        return train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "    except KeyError as e:\n",
    "        raise TargetMissingException from e"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Step 4: Model Training**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ModelTrainingException(ModelPipelineError):\n",
    "    def __init__(self):\n",
    "        super().__init__(\"The model was provided non allowed values.\")\n",
    "\n",
    "def train_model(X_train, y_train):\n",
    "    try:\n",
    "        model = LinearRegression()\n",
    "        model.fit(X_train, y_train)\n",
    "        return model\n",
    "    except ValueError as e:\n",
    "        raise ModelTrainingException from e"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Step 5: Evaluating the model**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "class ModelEvaluationException(ModelPipelineError):\n",
    "    def __init__(self):\n",
    "        super().__init__(\"The evaluation of the mode failed.\")\n",
    "\n",
    "def evaluate_model(model, X_test, y_test, predictions):\n",
    "    try:\n",
    "        predictions = model.predict(X_test)\n",
    "        mse = mean_squared_error(y_test, predictions)\n",
    "        print(f\"Model Mean Squared Error: {mse}\")\n",
    "    except Exception as e:\n",
    "        raise ModelEvaluationException from e "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Step 6: Running the Pipeline**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pipeline(datapath, predictions):\n",
    "    try:\n",
    "        # Step 1: Load the data\n",
    "        data = load_data(datapath)\n",
    "\n",
    "        # Step 2: Preprocess the data\n",
    "        X_train, X_test, y_train, y_test = preprocess_data(data)\n",
    "    \n",
    "        # Step 3: Train the model\n",
    "        model = train_model(X_train, y_train)\n",
    "\n",
    "        # Step 4: Evaluate the model\n",
    "        predictions = model.predict(X_test)\n",
    "        mse = mean_squared_error(y_test, predictions)\n",
    "        print(f\"Model Mean Squared Error: {mse}\")\n",
    "\n",
    "    except FileMissingException as e:\n",
    "        logger.exception(\"A problem was found while loading the data.\")\n",
    "        raise\n",
    "    except TargetMissingException as e:\n",
    "        logger.exception(\"A problem was found while preparing the data.\")\n",
    "        raise\n",
    "    except ModelTrainingException as e:\n",
    "        logger.exception(\"A problem was found while training the data.\")\n",
    "        raise\n",
    "    except ModelEvaluationException as e:\n",
    "        logger.exception(\"Error during model evaluation.\")\n",
    "        raise"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "---\n",
    "## 6. Exercise <a name=\"6\"></a>\n",
    "**Task**\n",
    "You are provided with a simple data science pipeline that loads data, validates it, preprocesses it, and trains a model. The pipeline currently does not have any exception handling. Your task is to:\n",
    " - Add exception handling to each step of the pipeline.\n",
    " - Use custom exceptions where appropriate.\n",
    " - Implement logging for all exceptions.\n",
    "\n",
    "**Initial Code**\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "import pandas as pd\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "def load_data(filepath):\n",
    "    data = pd.read_csv(filepath)\n",
    "    return data\n",
    "\n",
    "def validate_data(data):\n",
    "    if data.isnull().sum().sum() > 0:\n",
    "        print(\"Data contains missing values.\")\n",
    "\n",
    "def preprocess_data(data):\n",
    "    data['new_column'] = data['existing_column'] * 2\n",
    "    return data\n",
    "\n",
    "def train_model(data):\n",
    "    X = data[['new_column']]\n",
    "    y = data['target']\n",
    "    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "    model = LinearRegression()\n",
    "    model.fit(X_train, y_train)\n",
    "    return model\n",
    "\n",
    "def run_pipeline(filepath):\n",
    "    data = load_data(filepath)\n",
    "    validate_data(data)\n",
    "    data = preprocess_data(data)\n",
    "    model = train_model(data)\n",
    "    return model\n",
    "\n",
    "# Usage\n",
    "model = run_pipeline('data/raw/example.csv')\n",
    "print(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Requirements\n",
    "- Handle file not found errors in load_data.\n",
    "- Raise a custom exception for validation errors in validate_data.\n",
    "- Handle missing column errors in preprocess_data.\n",
    "- Handle any errors during model training in train_model.\n",
    "- Log all exceptions with appropriate severity levels.\n",
    "\n",
    "Write your solution here:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "def load_data(filepath):\n",
    "    data = pd.read_csv(filepath)\n",
    "    return data\n",
    "\n",
    "def validate_data(data):\n",
    "    if data.isnull().sum().sum() > 0:\n",
    "        print(\"Data contains missing values.\")\n",
    "\n",
    "def preprocess_data(data):\n",
    "    data['new_column'] = data['existing_column'] * 2\n",
    "    return data\n",
    "\n",
    "def train_model(data):\n",
    "    X = data[['new_column']]\n",
    "    y = data['target']\n",
    "    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "    model = LinearRegression()\n",
    "    model.fit(X_train, y_train)\n",
    "    return model\n",
    "\n",
    "def run_pipeline(filepath):\n",
    "    data = load_data(filepath)\n",
    "    validate_data(data)\n",
    "    data = preprocess_data(data)\n",
    "    model = train_model(data)\n",
    "    return model\n",
    "\n",
    "# Usage\n",
    "model = run_pipeline('data/raw/example.csv')\n",
    "print(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "**Solution**\n",
    "\n",
    "(careful, the solution is not correct, it's still being reviewed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import logging\n",
    "from sklearn.linear_model import LinearRegression\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "# Configure logging\n",
    "logging.basicConfig(level=logging.INFO)\n",
    "logger = logging.getLogger(__name__)\n",
    "\n",
    "class DataValidationError(Exception):\n",
    "    pass\n",
    "\n",
    "def load_data(filepath):\n",
    "    try:\n",
    "        data = pd.read_csv(filepath)\n",
    "        return data\n",
    "    except FileNotFoundError:\n",
    "        logger.error(f\"File not found: {filepath}\")\n",
    "        raise\n",
    "    except pd.errors.EmptyDataError:\n",
    "        logger.error(\"No data in file.\")\n",
    "        raise\n",
    "    except Exception as e:\n",
    "        logger.error(f\"An unexpected error occurred: {e}\")\n",
    "        raise\n",
    "\n",
    "def validate_data(data):\n",
    "    try:\n",
    "        if data.isnull().sum().sum() > 0:\n",
    "            raise DataValidationError(\"Data contains missing values.\")\n",
    "    except DataValidationError as e:\n",
    "        logger.warning(f\"Validation error: {e}\")\n",
    "        raise\n",
    "\n",
    "def preprocess_data(data):\n",
    "    try:\n",
    "        data['new_column'] = data['existing_column'] * 2\n",
    "        return data\n",
    "    except KeyError as e:\n",
    "        logger.error(f\"Missing column during preprocessing: {e}\")\n",
    "        raise DataValidationError(f\"Preprocessing error: {e}\")\n",
    "\n",
    "def train_model(data):\n",
    "    try:\n",
    "        X = data[['new_column']]\n",
    "        y = data['target']\n",
    "        X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\n",
    "        model = LinearRegression()\n",
    "        model.fit(X_train, y_train)\n",
    "        return model\n",
    "    except KeyError as e:\n",
    "        logger.error(f\"Missing target column: {e}\")\n",
    "        raise DataValidationError(f\"Training error: {e}\")\n",
    "    except Exception as e:\n",
    "        logger.error(f\"An error occurred during model training: {e}\")\n",
    "        raise\n",
    "\n",
    "def run_pipeline(filepath):\n",
    "    try:\n",
    "        data = load_data(filepath)\n",
    "        validate_data(data)\n",
    "        data = preprocess_data(data)\n",
    "        model = train_model(data)\n",
    "        return model\n",
    "    except DataValidationError as e:\n",
    "        logger.error(f\"Pipeline failed: {e}\")\n",
    "    except Exception as e:\n",
    "        logger.critical(f\"Critical error in pipeline: {e}\")\n",
    "\n",
    "# Usage\n",
    "try:\n",
    "    model = run_pipeline('data/raw/example.csv')\n",
    "    print(model)\n",
    "except Exception as e:\n",
    "    logger.critical(f\"Pipeline execution failed: {e}\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Intro_python",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}