from lib.enums import AthenaQueryNames
class TemplateError(Exception):
def __init__(self, query_name: AthenaQueryNames):
self.query_name = query_name
def __str__(self) -> str:
return f"No config found for query with name: {self.query_name}"
class TemplateFileNotFoundError(Exception):
def __init__(self, template_path: str):
self.template_path = template_path
def __str__(self) -> str:
return f"No template file found for template path: {self.template_path}"
class TableS3LocationNotFound(Exception):
def __init__(self, query_name: AthenaQueryNames):
self.query_name = query_name
def __str__(self) -> str:
return f"Table S3 location not found in query: {self.query_name}"
class AthenaAWSDisabledError(Exception):
pass
class AthenaQueryExecutionError(Exception):
execution_id: str
athena_error_message: str
def __init__(
self,
execution_id: str,
athena_error_message: str,
):
self.execution_id = execution_id
self.athena_error_message = athena_error_message
def __str__(self) -> str:
return self.athena_error_message
class HIVEPartitionExist(AthenaQueryExecutionError):
"""
Directory exist
This means most likely that unload is executed a second time
"""
class HIVECannotOpenSplit(AthenaQueryExecutionError):
"""
Problem with file opening, probably due to actualise data (replace s3 file)
Retry needed
"""
class HIVEMalformedData(AthenaQueryExecutionError):
"""
Malformed Parquet file with incompatible types
File needs to be fixed and query retried
"""
s3_file_path: str
def __init__(
self,
execution_id: str,
athena_error_message: str,
s3_file_path: str,
):
super().__init__(execution_id, athena_error_message)
self.s3_file_path = s3_file_path
class S3ObjectNotFoundError(Exception):
def __init__(self, bucket: str, key: str):
self.bucket = bucket
self.key = key
def __str__(self) -> str:
return f"Key '{self.key}' is not found in bucket '{self.bucket}'"
