modified project structure

Author: umang-clairvoyant

pyproject.toml

@@ -11,6 +11,9 @@ python = "^3.11"
 pydantic = "^1.10.6"
 sqlalchemy = "^2.0.7"
 pandas = "^1.5.3"
+loguru = "^0.6.0"
+openpyxl = "^3.1.2"
+pypika = "^0.48.9"
 
 
 [build-system]

src/app/api/end.py

@@ -0,0 +1,4 @@
+class AutoETLException(Exception):
+    """ Exception handler
+
+    """

src/app/etl_exceptions.py

@@ -0,0 +1,52 @@
+import logging
+import logging.config
+from datetime import datetime
+
+
+def setup_logging(log_dir: str):
+    """Load logging configuration"""
+
+    log_file_name = log_dir + '/' + 'minimal-app-' + datetime.now().strftime("%Y-%m-%d") + '.log'
+
+    loging_config = {
+        'version': 1,
+        'disable_existing_loggers': False,
+        'loggers': {
+            'root': {
+                'level': 'INFO',
+                'handlers': ['debug_console_handler', 'info_rotating_file_handler'],
+            },
+            'src': {
+                'level': 'DEBUG',
+                'propagate': False,
+                'handlers': ['info_rotating_file_handler', 'debug_console_handler'],
+            },
+        },
+        'handlers': {
+            'debug_console_handler': {
+                'level': 'DEBUG',
+                'formatter': 'console',
+                'class': 'logging.StreamHandler',
+                'stream': 'ext://sys.stdout',
+            },
+            'info_rotating_file_handler': {
+                'level': 'WARN',
+                'formatter': 'file',
+                'class': 'logging.handlers.RotatingFileHandler',
+                'filename': log_file_name,
+                'mode': 'a',
+                'maxBytes': 1048576,
+                'backupCount': 10
+            }
+        },
+        'formatters': {
+            'console': {
+                'format': '%(levelname)s:     %(name)s - %(message)s'
+            },
+            'file': {
+                'format': '%(asctime)s-%(levelname)s-%(name)s-%(process)d::%(module)s|%(lineno)s:: %(message)s'
+            },
+        },
+    }
+
+    logging.config.dictConfig(loging_config)

src/app/logger.py

@@ -0,0 +1 @@
+from .parser import ConfigParser, MetadataParser

src/app/services/config_parser/__init__.py

@@ -0,0 +1,98 @@
+import json
+import logging
+from typing import Any
+
+import pandas as pd
+from pydantic.dataclasses import dataclass
+
+from app.etl_exceptions import AutoETLException
+from app.utils import is_valid_file
+
+logger = logging.getLogger(__name__)
+
+sheet_names = ["target_table", "joins_and_filters"]
+
+
+@dataclass
+class MetadataParser:
+    metadata_file_path: str
+
+    def validate_file(self) -> pd.ExcelFile:
+        """method to validate the metadata excel file
+
+        Raises:
+            AutoETLException: Exception if sheet names not found in the file
+
+        Returns:
+            ExcelFile
+        """
+        logger.info("Validating the metadata file")
+        if not is_valid_file(self.metadata_file_path):
+            logger.error("File path not correct - %s", self.metadata_file_path)
+            raise AutoETLException(
+                f"File path not correct - {self.metadata_file_path}")
+        meta_xls = pd.ExcelFile(self.metadata_file_path)
+        current_sheets = meta_xls.sheet_names
+
+        if not all(sheet in current_sheets for sheet in sheet_names):
+            logger.error(
+                "Sheet names not set properly in metadata excel file")
+            raise AutoETLException(
+                "Sheet names not set properly in metadata excel file")
+        return meta_xls
+
+    @classmethod
+    def get_metadata_parser(cls,
+                            file_path: str) -> 'MetadataParser':
+        """method to get the metadata parser object
+
+        Args:
+            file_path (str): path of the metadata file
+
+        Returns:
+            MetadataParser
+        """
+        return MetadataParser(file_path)
+
+
+@dataclass
+class ConfigParser:
+    config_file_path: str
+
+    def validate_file(self) -> Any:
+        """method to validate the config Json file
+
+        Raises:
+            AutoETLException: Exception if required configs are missing
+
+        Returns:
+            Dictionary
+        """
+        logger.info("Validating the config file")
+        if not is_valid_file(self.config_file_path):
+            logger.error("File path not correct - %s", self.config_file_path)
+            raise AutoETLException(
+                f"File path not correct - {self.config_file_path}")
+
+        with open(self.config_file_path) as fp:
+            config_json = json.load(fp)
+
+        if config_json.get('target', None) is None:
+            logger.error(
+                "Target system not set properly in config file")
+            raise AutoETLException(
+                "Target system not set properly in config file")
+        return config_json
+
+    @classmethod
+    def get_config_parser(cls,
+                          file_path: str) -> 'ConfigParser':
+        """method to get the config parser object
+
+        Args:
+            file_path (str): path of the config file
+
+        Returns:
+            ConfigParser
+        """
+        return ConfigParser(file_path)

src/app/services/config_parser/metadata_parser.py

@@ -0,0 +1,2 @@
+from .dialact import RedshiftDialact
+from .query_builder import QueryBuilder

src/app/services/config_parser/parser.py

@@ -0,0 +1,83 @@
+import logging
+from typing import Any, Dict, List
+
+from pydantic.dataclasses import dataclass
+from pypika import Query, Schema, Table
+from pypika.enums import Dialects, JoinType
+from pypika.queries import QueryBuilder
+from pypika.utils import builder
+
+from app.services.query_builder.join_queries import JoinQuery
+
+logger = logging.getLogger(__name__)
+
+
+class RedshiftQuery(Query):
+    """
+    Query class for AWS Redshift
+    """
+
+    @classmethod
+    def _builder(cls, **kwargs: Any) -> "RedshiftQueryBuilder":
+        return RedshiftQueryBuilder(**kwargs)
+
+
+class RedshiftQueryBuilder(QueryBuilder):
+    QUERY_CLS = RedshiftQuery
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(dialect=Dialects.REDSHIFT, **kwargs)
+
+    @builder
+    def join(
+        self, item: Table, how: JoinType = JoinType.inner
+    ) -> "JoinQuery":
+        if isinstance(item, Table):
+            return JoinQuery(self, item, how, label="table")
+
+        raise ValueError(f"Cannot join on type {type(item)}")
+
+    def inner_join(self, item: Table) -> "JoinQuery":
+        return self.join(item, JoinType.inner)
+
+    def left_join(self, item: Table) -> "JoinQuery":
+        return self.join(item, JoinType.left)
+
+
+@dataclass()
+class RedshiftDialact:
+    target_table_conf: List[Dict]
+    joins_and_filers_conf: Dict[str, Dict]
+
+    def build(self) -> None:
+        """Method to trrigger Redshift query builder
+        """
+        logger.info('building Redshift query from the mappings file')
+
+        _query = RedshiftQuery()
+        for _index in self.joins_and_filers_conf:
+
+            _map = self.joins_and_filers_conf[_index]
+
+            if int(_index) == 0:
+                schema1, schema2 = Schema(_map['driving_table'].split(
+                    '.')[0]), Schema(_map['reference_table'].split('.')[0])
+
+                table1, table2 = Table(_map['driving_table'].split('.')[1],
+                                       schema=schema1, alias=_map['driving_table_alias']), \
+                    Table(_map['reference_table'].split('.')[1],
+                          schema=schema2, alias=_map['reference_table_alias'])
+
+                _query = _query.from_(table1).inner_join(  # type: ignore
+                    table2).on(_map['join_condition'])
+            else:
+                schema2 = Schema(_map['reference_table'].split(
+                    '.')[0])
+
+                table2 = Table(_map['reference_table'].split('.')[1],
+                               schema=schema2, alias=_map['driving_table_alias'])
+
+                _query = _query.left_join(table2).on(
+                    _map['join_condition'])
+
+        print(_query.select('*'))

src/app/services/io/__init__.py

@@ -0,0 +1,53 @@
+import logging
+from typing import Any, Optional, Union
+
+from pydantic.dataclasses import dataclass
+from pypika import Criterion, Table
+from pypika.enums import JoinType
+from pypika.queries import Join, Joiner, QueryBuilder
+from pypika.terms import Term
+
+from app.etl_exceptions import AutoETLException
+
+logger = logging.getLogger(__name__)
+
+
+class Config:
+    arbitrary_types_allowed = True
+
+
+@dataclass(config=Config)
+class JoinQuery(Joiner):
+    query: QueryBuilder
+    item: Table
+    how: JoinType
+    label: str
+
+    def __post_init__(self):
+        super().__init__(self.query, self.item, self.how, self.label)
+
+    def on(self, criterion: Criterion, collate: Optional[str] = None) -> QueryBuilder:
+        if criterion is None:
+            logger.error(
+                "Parameter 'criterion' is required for a %s JOIN but was not supplied.", self.label)
+            raise AutoETLException(
+                f"Parameter 'criterion' is required for a {self.label} JOIN but was not supplied.")
+
+        self.query.do_join(JoinQueryOn(
+            self.item, self.how, criterion, collate))  # type: ignore
+        return self.query
+
+
+@dataclass(config=Config)
+class JoinQueryOn(Join):
+    def __init__(self, item: Term, how: JoinType, criteria: Union[QueryBuilder, str],
+                 collate: Optional[str] = None) -> None:
+        super().__init__(item, how)
+        self.criterion = criteria
+        self.collate = collate
+
+    def get_sql(self, **kwargs: Any) -> str:
+        join_sql = super().get_sql(**kwargs)
+        criteria = self.criterion if isinstance(
+            self.criterion, str) else self.criterion.get_sql(subquery=True, **kwargs)
+        return f"{join_sql} ON {criteria} {f' COLLATE {self.collate}' if self.collate else ''}"

src/app/services/io/databases/__init__.py

@@ -0,0 +1,61 @@
+import dataclasses
+import logging
+import os
+
+from pydantic.dataclasses import dataclass
+
+from app.etl_exceptions import AutoETLException
+from app.services.config_parser import ConfigParser, MetadataParser
+from app.services.query_builder import RedshiftDialact
+from app.utils import excel_to_json, validate_joins_mapping
+
+logger = logging.getLogger(__name__)
+
+
+class Config:
+    arbitrary_types_allowed = True
+
+
+@dataclass(config=Config)
+class QueryBuilder:
+    metadata_file_path: str
+    config_file_path: str
+    metadata_parser: MetadataParser = dataclasses.field(init=False)
+    config_parser: ConfigParser = dataclasses.field(init=False)
+
+    def __post_init__(self):
+        self.metadata_parser: MetadataParser = MetadataParser.get_metadata_parser(
+            self.metadata_file_path)
+        self.config_parser: ConfigParser = ConfigParser.get_config_parser(
+            self.config_file_path)
+
+    def run(self) -> None:
+        """ Method to trigger the build
+        """
+        logger.info("Initialising Auto ETL")
+        logger.info("Found metadata file - %s",
+                    os.path.basename(self.metadata_file_path))
+        logger.info("Found config file - %s",
+                    os.path.basename(self.config_file_path))
+
+        meta_xls = self.metadata_parser.validate_file()
+        _config = self.config_parser.validate_file()
+
+        logger.info("building query from meta_file -> %s",
+                    {self.metadata_file_path})
+
+        target_table_json = excel_to_json(meta_xls, 'target_table', 'records')
+        joins_and_filters = excel_to_json(
+            meta_xls, 'joins_and_filters', 'index')
+
+        validate_joins_mapping(joins_and_filters)
+
+        match _config['target']:
+            case 'redshift':
+                RedshiftDialact(target_table_json, joins_and_filters).build()
+
+            case _:
+                logger.error(
+                    "Target system - %s not supported yet.", _config.get('target', 'None'))
+                raise AutoETLException(
+                    f"Target system - {_config['target']} not supported yet.")