Commit e6365034 authored by Niall McDaid's avatar Niall McDaid
Browse files

restructure files into layers

parent e1b85e06
from typing import Any, Dict, List, Union
from .file_loader import IFileLoader, JsonLoader
from .reflection_utilities import ReflectionHelper
class Configuration:
"""
Class to hold metadata for LAS ingestion
"""
def __init__(self, file_loader: IFileLoader, path: str) -> None:
"""
Load JSON configuration using the file loader and create a new instance of a Configuration
:param IFileLoader file_loader: The file loader instance to user
:param str path: The full path and filename of the configuration file.
"""
json_parser = JsonLoader(file_loader)
self._config = json_parser.load(path)
def get_recursive(self, qualified_attribute_name: List[str]) -> str:
"""
Gets an item specified by its qualified attribute name - eg ["legal", "legaltags"] will get config[legal][legaltags].
:param str qualified_attribute_name: The exception message
return: the base url for API calls
rtype: str
"""
return ReflectionHelper.getattr_recursive(self._config, qualified_attribute_name)
@property
def base_url(self) -> str:
"""
Gets the base url.
return: the base url for API calls
rtype: str
"""
return self._config.get("base_url")
@property
def data_partition_id(self) -> str:
"""
Gets the data partition Id.
return: the data partition id
rtype: str
"""
return self._config.get("data_partition_id")
@property
def wellbore_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the wellbore_mapping.
return: the wellbore_mapping
rtype: Dict[str, Any]
"""
return self._config.get("wellbore_mapping")
@property
def welllog_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the welllog_mapping.
return: the welllog_mapping
rtype: Dict[str, Any]
"""
return self._config.get("welllog_mapping")
from typing import Any, Dict, List, Union
from .file_loader import IFileLoader, JsonLoader
from .reflection_utilities import ReflectionHelper
class Configuration:
"""
Class to hold metadata for LAS ingestion
"""
def __init__(self, file_loader: IFileLoader, path: str) -> None:
"""
Load JSON configuration using the file loader and create a new instance of a Configuration
:param IFileLoader file_loader: The file loader instance to user
:param str path: The full path and filename of the configuration file.
"""
json_parser = JsonLoader(file_loader)
self._config = json_parser.load(path)
def get_recursive(self, qualified_attribute_name: List[str]) -> str:
"""
Gets an item specified by its qualified attribute name - eg ["legal", "legaltags"] will get config[legal][legaltags].
:param str qualified_attribute_name: The exception message
return: the base url for API calls
rtype: str
"""
return ReflectionHelper.getattr_recursive(self._config, qualified_attribute_name)
@property
def base_url(self) -> str:
"""
Gets the base url.
return: the base url for API calls
rtype: str
"""
return self._config.get("base_url")
@property
def data_partition_id(self) -> str:
"""
Gets the data partition Id.
return: the data partition id
rtype: str
"""
return self._config.get("data_partition_id")
@property
def wellbore_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the wellbore_mapping.
return: the wellbore_mapping
rtype: Dict[str, Any]
"""
return self._config.get("wellbore_mapping")
@property
def welllog_mapping(self) -> Union[Dict[str, Any], None]:
"""
Gets the welllog_mapping.
return: the welllog_mapping
rtype: Dict[str, Any]
"""
return self._config.get("welllog_mapping")
from typing import List
class WbdutilException(Exception):
"""
Base class for exceptions raised within Wdutil
"""
def __init__(self, *args):
"""
Create a new instance of a WdutilException
:param *args object: The Exception arguments
"""
super().__init__(args)
class WbdutilValidationException(Exception):
"""
General validation exception raised within Wdutil
"""
def __init__(self, error_messages: List[str], *args):
"""
Create a new instance of a WdutilValidationException
:param error_messages List[str]: The validation errors
:param *args object: The Exception arguments
"""
self._error_messages = error_messages
super().__init__(args)
@property
def error_messages(self) -> List[str]:
"""
Gets the list of validation errors
return: the validation errors
rtype: List[str]
"""
return self._error_messages
class WbdutilAttributeError(WbdutilException):
"""
Common exception class for attribute errors raised in the Wbdutil
"""
def __init__(self, message: str, inner_exception: Exception, *args):
"""
Create a new instance of a ExtendedPropertiesAttributeError
:param str message: The exception message
:param inner_exception Exception: The inner exception
:param args object: The Exception arguments
"""
self.message = message
self.inner_exception = inner_exception
super().__init__(args)
class WbdutilValueError(Exception):
def __init__(self, *args):
"""
Create a new instance of a WbdutilValueError
:param *args object: The Exception arguments
"""
super().__init__(args)
from typing import List
class WbdutilException(Exception):
"""
Base class for exceptions raised within Wdutil
"""
def __init__(self, *args):
"""
Create a new instance of a WdutilException
:param *args object: The Exception arguments
"""
super().__init__(args)
class WbdutilValidationException(Exception):
"""
General validation exception raised within Wdutil
"""
def __init__(self, error_messages: List[str], *args):
"""
Create a new instance of a WdutilValidationException
:param error_messages List[str]: The validation errors
:param *args object: The Exception arguments
"""
self._error_messages = error_messages
super().__init__(args)
@property
def error_messages(self) -> List[str]:
"""
Gets the list of validation errors
return: the validation errors
rtype: List[str]
"""
return self._error_messages
class WbdutilAttributeError(WbdutilException):
"""
Common exception class for attribute errors raised in the Wbdutil
"""
def __init__(self, message: str, inner_exception: Exception, *args):
"""
Create a new instance of a ExtendedPropertiesAttributeError
:param str message: The exception message
:param inner_exception Exception: The inner exception
:param args object: The Exception arguments
"""
self.message = message
self.inner_exception = inner_exception
super().__init__(args)
class WbdutilValueError(Exception):
def __init__(self, *args):
"""
Create a new instance of a WbdutilValueError
:param *args object: The Exception arguments
"""
super().__init__(args)
from typing import Any, Dict, List
from knack.log import get_logger
from .exceptions import WbdutilAttributeError, WbdutilValueError
logger = get_logger(__name__)
class ReflectionHelper:
def getattr_recursive(source: any, source_field_heirachy: List[str]) -> any:
"""
Get the specified field from the source object in a recursive manner.
:param any destination: The source object
:param List[str] source_field_heirachy: List of fields, from lowest to the deepest in the heirachy.
:return: The data value.
:rtype: any
"""
field_name = source_field_heirachy.pop(0)
try:
if isinstance(source, dict):
subfield = source[field_name]
else:
subfield = getattr(source, field_name)
except (KeyError, AttributeError):
if field_name.endswith(']') and '[' in field_name:
# probably a reference to a list item
logger.debug(f"Attempting to parse the mapping field {field_name} as an array")
subfield = ReflectionHelper._getattr_list(source, field_name)
else:
message = f"The attribute '{field_name}' was not found in the source data. An empty value for this field will be returned."
logger.error(message)
return None
if len(source_field_heirachy) == 0:
return subfield
else:
return ReflectionHelper.getattr_recursive(subfield, source_field_heirachy)
def _getattr_list(source: any, source_field_name: str) -> any:
try:
segments = source_field_name.split('[')
if isinstance(source, dict):
subfield = source[segments.pop(0)]
else:
subfield = getattr(source, segments.pop(0))
for ind in segments:
subfield = subfield[int(ind.rstrip(']'))]
return subfield
except (KeyError, AttributeError, ValueError) as ex:
message = f"The attribute '{source_field_name}' could not be parsed as an array"
logger.error(message)
raise WbdutilAttributeError(message, ex)
def setattr_recursive(value: any, destination: Dict[str, Any], destination_field_heirachy: List[str]) -> None:
"""
Set the specified field to the given value, recursively create member objects where needed. Update in place.
:param any value: The data value
:param any destination: The destination object
:param List[str] destination_field_heirachy: List of fields in the hierarchy, ordered from top to bottom.
"""
if destination is None:
message = "The destination dictionary must not be null"
logger.critical(message)
raise WbdutilValueError(message)
field_name = destination_field_heirachy.pop(0)
if len(destination_field_heirachy) == 0:
destination[field_name] = value
return destination
if field_name not in destination:
destination[field_name] = {}
ReflectionHelper.setattr_recursive(value, destination[field_name], destination_field_heirachy)
from typing import Any, Dict, List
from knack.log import get_logger
from .exceptions import WbdutilAttributeError, WbdutilValueError
logger = get_logger(__name__)
class ReflectionHelper:
def getattr_recursive(source: any, source_field_heirachy: List[str]) -> any:
"""
Get the specified field from the source object in a recursive manner.
:param any destination: The source object
:param List[str] source_field_heirachy: List of fields, from lowest to the deepest in the heirachy.
:return: The data value.
:rtype: any
"""
field_name = source_field_heirachy.pop(0)
try:
if isinstance(source, dict):
subfield = source[field_name]
else:
subfield = getattr(source, field_name)
except (KeyError, AttributeError):
if field_name.endswith(']') and '[' in field_name:
# probably a reference to a list item
logger.debug(f"Attempting to parse the mapping field {field_name} as an array")
subfield = ReflectionHelper._getattr_list(source, field_name)
else:
message = f"The attribute '{field_name}' was not found in the source data. An empty value for this field will be returned."
logger.error(message)
return None
if len(source_field_heirachy) == 0:
return subfield
else:
return ReflectionHelper.getattr_recursive(subfield, source_field_heirachy)
def _getattr_list(source: any, source_field_name: str) -> any:
try:
segments = source_field_name.split('[')
if isinstance(source, dict):
subfield = source[segments.pop(0)]
else:
subfield = getattr(source, segments.pop(0))
for ind in segments:
subfield = subfield[int(ind.rstrip(']'))]
return subfield
except (KeyError, AttributeError, ValueError) as ex:
message = f"The attribute '{source_field_name}' could not be parsed as an array"
logger.error(message)
raise WbdutilAttributeError(message, ex)
def setattr_recursive(value: any, destination: Dict[str, Any], destination_field_heirachy: List[str]) -> None:
"""
Set the specified field to the given value, recursively create member objects where needed. Update in place.
:param any value: The data value
:param any destination: The destination object
:param List[str] destination_field_heirachy: List of fields in the hierarchy, ordered from top to bottom.
"""
if destination is None:
message = "The destination dictionary must not be null"
logger.critical(message)
raise WbdutilValueError(message)
field_name = destination_field_heirachy.pop(0)
if len(destination_field_heirachy) == 0:
destination[field_name] = value
return destination
if field_name not in destination:
destination[field_name] = {}
ReflectionHelper.setattr_recursive(value, destination[field_name], destination_field_heirachy)
from abc import ABC, abstractmethod
from typing import Any, Dict
from knack.log import get_logger
from .configuration import Configuration
from .exceptions import WbdutilAttributeError
from .reflection_utilities import ReflectionHelper
logger = get_logger(__name__)
class IPropertyMappingLoader(ABC):
@property
@abstractmethod
def mapping(self) -> Dict[str, Any]:
pass
@property
@abstractmethod
def kind(self) -> str:
pass
class DictionaryMappingLoader(IPropertyMappingLoader):
def __init__(self, base_mapping: Dict[str, Any]) -> None:
self._base_mapping = base_mapping
@property
def mapping(self) -> Dict[str, Any]:
"""
Gets the mapping dictionary.
return:The mapping
rtype: Dict[str, any]
"""
return self._base_mapping["mapping"]
@property
def kind(self) -> str:
"""
Gets the kind for the mapping
return: the kind string
rtype: str
"""
return self._base_mapping["kind"]
class PropertyMapper:
"""
Class to handle the mapping of one data type to another using a dictionary of mappings
"""
_mapping_functions: Any
_config: Configuration
_kind: str
_mapping: Dict[str, Any]
def __init__(self, loader: IPropertyMappingLoader, config: Configuration, mapping_functions: any) -> None:
"""
Create a new instance of a PropertyMapper
:param IPropertyMappingLoader loader: A IPropertyMappingLoader loader instance that
contains the mapping between data types.
:param Configuration config: The application configuration
:param any mapping_functions: An instance of a class that contains the available mapping functions
"""
self._mapping = loader.mapping
self._kind = loader.kind
self._config = config
self._mapping_functions = mapping_functions
def remap_data_with_kind(self, source: any) -> Dict[str, Any]:
"""
Remap the data in the source object using the mapping dictionary to a new object. Adds the kind field extracted from the config
:param any source: The source object
:return: A new data object that contains the remapped data
:rtype: Dict[str, Any]
"""
output = self.remap_data(source)
output["kind"] = self._kind
return output
def remap_data(self, source: any) -> Dict[str, Any]:
"""
Remap the data in the source object using the mapping dictionary to a new object
:param any source: The source object
:return: A new data object that contains the remapped data
:rtype: Dict[str, Any]
"""
return self._remap_data(source, self._mapping)
def _remap_data(self, source: any, mapping: Dict[str, Any]) -> Dict[str, Any]:
extensible = {}
# iterate through the mapping dictionary remapping data as we go
for target_field, source_field in mapping.items():
source_data = self._get_source_attribute_value(source, source_field)
ReflectionHelper.setattr_recursive(source_data, extensible, target_field.split('.'))
return extensible
def _get_source_attribute_value(self, source, source_field):
if isinstance(source_field, dict):
source_data = self._process_complex_mapping(source, source_field)
else:
source_field_list = source_field.split('.')
if source_field_list[0] == "CONFIGURATION":
# get data from config
del source_field_list[0]
source_data = self._config.get_recursive(source_field_list)
else:
# Get data from source
source_data = ReflectionHelper.getattr_recursive(source, source_field_list)
return source_data
def _process_complex_mapping(self, source: any, source_field: Dict[str, Any]) -> any:
mapping_type = source_field.get("type")
if mapping_type is None:
raise ValueError("The mapping type must be specified. Supported types are: function, array")
if mapping_type == "function":
try:
function_name = source_field.get("function")
if function_name is None:
raise ValueError("The function name must be given in the configuration file")
argument_names = source_field.get("args", [])
if not isinstance(argument_names, list):
message = f"The mapping function arguments must be a list but was '{type(argument_names)}'."
logger.error(message)
raise WbdutilAttributeError(message, None)
if len(argument_names) > 0 and any(not isinstance(a, str) for a in argument_names):
message = "All of the mapping function arguments must be strings."
logger.error(message)
raise WbdutilAttributeError(message, None)
arguments = [self._get_source_attribute_value(source, a) for a in argument_names]
function_to_call = getattr(self._mapping_functions, function_name)
return function_to_call(*arguments)
except AttributeError as ex:
message = f"The function '{function_name}' was not found in the provided mapping functions"
logger.error(message)
raise WbdutilAttributeError(message, ex)
elif mapping_type == "array":
source_array_name = source_field.get("source")
source_array = self._get_source_attribute_value(source, source_array_name)
submapping = source_field.get("mapping", {})
remapped_array = []
for item in source_array: