ErrorHandler.py

Go to the documentation of this file.

import os
from datetime import datetime as dt
from pymongo import MongoClient
import ssl
from dotenv import load_dotenv, dotenv_values
from MongoConnection import mongo_connection
load_dotenv()
 
class ErrorHandler:
    """Updates file info in our database with the error found in it (if it's applicable)
    """
 
    def handle(self, e, message, file_db_id):
        """Add status_cause and error_message fields to file info  
 
        Args:
            e (str): error found
            message (str): error message generated by python
            file_db_id (str): file id assigned in mongodb
        Returns: error_cause (str)
        """
        mongo_conn = mongo_connection()
        collection = mongo_conn.mongo_conn_snapshots()
        error_cause = self.get_human_error(e)
        collection.update_one({"file_db_id":file_db_id},{"$set": {"status": "failed", "status_cause": error_cause, "error_message": message, "updated_at": dt.now()}})
        # collection.update_one({"file_db_id":file_db_id},{"$set": {"updated_at": dt.now().replace(hour=0, minute=0, second=0, microsecond=0)}})
        return error_cause
    
    def get_human_error(self, e):
        """Gets a message error according to king of issue
 
        Args:
            e (str): error found
        Returns: error (str)
        """
        error_type = type(e)
        if str(error_type) == "<class 'botocore.exceptions.ClientError'>":
            return ("S3: File does not exist. Are you sure this is the correct file")
        elif str(error_type) == "<class 'botocore.exceptions.PartialCredentialsError'>":
            return ("S3: Partial credentials found in explicit. aws_access_key_id or aws_secret_access_key no founded")
        elif str(e) == "Path invalid.":
            return ("S3: Input path does not exist in s3 Bucket. Are you sure this is the correct path?")
        elif str(error_type) == "<class 'pymongo.errors.ServerSelectionTimeoutError'>":
            return ("MDB: Saving the information is taking too long. Connection refused")
        elif str(error_type) == "<class 'IndexError'>":
            return ("MDB: No formats matches with your file")
        elif str(e) == "Error in currency. Execution aborted":
            return ("PD: Currency not found in your file")
        elif str(e) == "Validation Error. Execution aborted":
            return ("PD: Something was wrong. Data validation failed.")
        elif str(error_type) == "<class 'NameError'>":
            return ("Python: There is a word not defined")
        elif str(error_type) == "<class 'KeyError'>":
            return ("Python: There is a wrong word")
        elif "Expecting value: line 1 column" in str(e):
            return ("Maybe there is a malformed format")
        elif "cannot unpack non-iterable" in str(e):
            return "Maybe something is wrong in headers" 
        return str(e)
 
# class DataBuildFailed(Exception):
# #     # def __init__(self):
# #     #     return None
#     # def __str__(): 
#     print("Data Build Failed. Please review data")
 
class FileNotSupported(Exception):
    """ 
    It's launched when a not allowed file is ingested 
    """
    def __init__(self, message="File extension is not supported. Verify if it is the correct file"):
        self.message = message
        super().__init__(self.message)
 
class FileNotLoaded(Exception):
    """ 
    It's launched when a file could not be loaded or it has a undefined encoding
    """
    def __init__(self, message="File couldn't be loaded. Verify that file is an accepted format"):
        self.message = message
        super().__init__(self.message)
 
class CurrencyNotFound(Exception):
    """ 
    It's launched when a file contains one or more not defined currencies
    """
    def __init__(self, message="There are currencies not found in your file. Please check currencies"):
        self.message = message
        super().__init__(self.message)
 
class ValidationFailed(Exception):
    """ 
    It's launched when something's wrong in the data validation procedure
    """
    def __init__(self, message="Something's wrong in validation step. Verify your data"):
        self.message = message
        super().__init__(self.message)