Merge pull request #1 from PageUpPeopleOrg/feature/OSC-967-ability-to-start-new-execution

[OSC-967] - Add ability to start new data pipeline execution

Author: ChintanRaval

.gitignore

@@ -102,3 +102,9 @@ venv.bak/
 
 # mypy
 .mypy_cache/
+
+# Visual Studio Code
+.vscode/
+
+# JetBrains
+.idea/

README.md

@@ -1 +1,119 @@
-# model-change-detector
+# Model Change Detector
+
+## About
+
+A utility that detects changes in models.  
+
+## Usage
+
+```commandline
+py mcd.py <command> <db-connection-string> [--help] [--log-level]
+```
+
+- `command` is the function to be performed by the utility. The currently supported values are 
+  - `START`: Marks the start of a new execution by creating a record for the same in the given database and returns an ID of the new execution.
+- `db-connection-string` is a [PostgreSQL Db Connection String](http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html#module-sqlalchemy.dialects.postgresql.psycopg2) of the format `postgresql+psycopg2://user:password@host:port/dbname`
+
+### As a script
+
+- Use a local isolated/virtual python environment for this project
+- Install project dependencies
+- `py mcd.py <command> <db-connection-string> [--help] [--log-level]`
+
+_Windows example:_
+
+```commandline
+py -m venv new-env --clear
+new-env\scripts\activate
+
+py -m pip install -r requirements.txt
+
+py mcd.py START postgresql+psycopg2://user:password@host:port/dbname
+```
+
+### As a package
+
+- Use/create an empty directory
+- Use a local isolated/virtual python environment for this project
+- [Install](https://pip.pypa.io/en/stable/reference/pip_install/#editable-installs) this package
+- `py -m mcd <command> <db-connection-string> [--help] [--log-level]`
+
+_Windows example:_
+
+```commandline
+mkdir new-dir
+cd new-dir
+
+py -m venv new-env --clear
+new-env\scripts\activate
+
+pip install -e path\to\model-change-detector
+
+py -m mcd START postgresql+psycopg2://user:password@host:port/dbname
+```
+
+## Setup
+
+1. Install pre-requisites
+2. Use a local isolated/virtual python environment for this project
+3. Install project dependencies
+4. Develop and test code changes 
+5. Once done, deactivate the virtual environment
+
+### Install pre-requisites
+
+#### Python 3
+
+Install from [here](https://www.python.org/) _(or your choice of safe software provider)_. During installation, choose the option to _Add to PATH_ and _Custom Installation_ so you can provide the install path to not be the default `C:\Program Files\Python37\` and be outside the `Program Files` directory to say, `C:\Python37\`. This is just a suggestion since there have been issues with updating key python packages once Python is installed within `Program Files` on Windows.
+
+Verify your installation by running the below commands.
+
+```powershell
+py --version
+python --version
+pip --version
+```
+
+If you end up with multiple accidental/purposeful python installations, use the below in Windows Commandline to figure out where the executables are located.
+
+```cmd
+where py
+where python
+where pip
+```
+
+### Use a local isolated/virtual python environment for this project
+
+`py -m venv /path/to/new/virtual/environment` _e.g._ `py -m venv new-env`
+
+If you build with `--system-site-packages` directory, your virtual environment will be allowed access to packages from your global site-packages directory. Although, if you want isolation from the global system, do not use this flag. Once you've created a new environment, you need to activate the same.
+
+On Windows:
+
+`path\to\environment\scripts\activate` _e.g._ `new-env\scripts\activate`
+
+On Linux / Mac OS
+
+ `source path/to/environment/bin/activate` _e.g._ `source new-env/bin/activate`
+
+You should see the name of your virtual environment in brackets on your terminal line, e.g.: 
+```
+C:\path\to\working\dir: new-env\scripts\activate
+(new-env) C:\path\to\working\dir: _
+```
+Any python commands you use will now, work within your virtual environment only.
+
+### Install project dependencies
+
+```powershell
+pip install -r requirements.txt
+```
+
+### Deactivate the virtual environment
+
+Once done, deactivate the virtual environment with a simple `decativate` command, e.g.: 
+
+```commandline
+(new-env) C:\path\to\working\dir: deactivate
+C:\path\to\working\dir: _
+```

mcd.py

@@ -0,0 +1,4 @@
+from modules.ModelChangeDetector import ModelChangeDetector
+
+if __name__ == "__main__":
+    ModelChangeDetector().main()

modules/BaseObject.py

@@ -0,0 +1,7 @@
+import logging
+
+
+class BaseObject(object):
+    def __init__(self, logger=None):
+        self.logger = logger or logging.getLogger(self.__module__ + '.' + self.__class__.__qualname__)
+        self.logger.debug(self.logger)

modules/DataPipelineExecutionEntity.py

@@ -0,0 +1,47 @@
+from sqlalchemy import Column, DateTime, Integer, String
+from sqlalchemy.sql import func
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.dialects.postgresql import UUID
+import uuid
+from modules.Shared import Constants
+
+Base = declarative_base()
+
+
+class DataPipelineExecutionEntity(Base):
+
+    __tablename__ = 'data_pipeline_execution'
+    __table_args__ = {'schema': Constants.DATA_PIPELINE_EXECUTION_SCHEMA_NAME}
+
+    id = Column('id',
+                UUID(as_uuid=True),
+                primary_key=True,
+                default=uuid.uuid4())
+
+    created_on = Column('created_on',
+                        DateTime(timezone=True),
+                        nullable=False,
+                        server_default=func.now())
+
+    last_updated_on = Column('last_updated_on',
+                             DateTime(timezone=True),
+                             nullable=False,
+                             server_default=func.now(),
+                             onupdate=func.now())
+
+    # unused for now, for future use
+    execution_time_ms = Column('execution_time_ms',
+                               Integer,
+                               nullable=True)
+
+    status = Column('status',
+                    String(50),
+                    nullable=False,
+                    server_default=str(Constants.DataPipelineExecutionStatus.STARTED))
+
+    def __str__(self):
+        return f'id={self.id}, ' \
+               f'created_on={self.created_on}, ' \
+               f'last_updated_on={self.last_updated_on}, ' \
+               f'execution_time_ms={self.execution_time_ms}, ' \
+               f'status={self.status}.'

modules/DataPipelineExecutionRepository.py

@@ -0,0 +1,30 @@
+from sqlalchemy import desc
+from modules.DataPipelineExecutionEntity import DataPipelineExecutionEntity, Base
+from modules.BaseObject import BaseObject
+from modules.Shared import Constants
+
+
+class DataPipelineExecutionRepository(BaseObject):
+    def __init__(self, session_maker, logger=None):
+        super().__init__(logger)
+        self.session_maker = session_maker
+
+    def create_schema(self, engine):
+        engine.execute(f'CREATE SCHEMA IF NOT EXISTS {Constants.DATA_PIPELINE_EXECUTION_SCHEMA_NAME}')
+        Base.metadata.create_all(engine)
+
+    def start_new(self):
+        session = self.session_maker()
+        data_pipeline_execution = DataPipelineExecutionEntity()
+        session.add(data_pipeline_execution)
+        session.commit()
+        return data_pipeline_execution
+
+    def get_last_successful_data_load_execution(self):
+        session = self.session_maker()
+        return session.query(DataPipelineExecutionEntity)\
+            .filter_by(status=Constants.DataPipelineExecutionStatus.COMPLETED_SUCCESSFULLY)\
+            .order_by(desc(DataPipelineExecutionEntity.created_on))\
+            .order_by(desc(DataPipelineExecutionEntity.last_updated_on))\
+            .first()
+

modules/ModelChangeDetector.py

@@ -0,0 +1,53 @@
+import argparse
+import logging
+from modules import Shared
+from modules.Shared import Constants
+from modules.commands import Commands
+from modules.commands.CommandFactory import CommandFactory
+from modules.BaseObject import BaseObject
+
+
+class ModelChangeDetector(BaseObject):
+
+    _commandNames = [Commands.get_name(Commands.START)]
+
+    def __init__(self, logger=None):
+        self.args = self.get_arguments()
+        Shared.configure_root_logger(self.args.log_level)
+
+        super().__init__(logger)
+
+        self.logger.debug(self.args)
+        self.logger.debug(f'args.log_level = {self.args.log_level} = {logging.getLevelName(self.args.log_level)}')
+
+        self.command_factory = CommandFactory()
+
+    def main(self):
+        command_executor = self.command_factory.create_command(self.args.command, self.args.db_connection_string)
+        command_executor.execute()
+
+    def get_arguments(self):
+        parser = argparse.ArgumentParser(description=Constants.APP_NAME,
+                                         parents=[Shared.get_default_arguments()])
+
+        parser.add_argument('command',
+                            type=self.get_command_value_from_name,
+                            help=f'choose from {", ".join(self._commandNames)}, more coming soon..')
+
+        parser.add_argument('db_connection_string',
+                            metavar='db-connection-string',
+                            help='provide in PostgreSQL & Psycopg format, '
+                                 'postgresql+psycopg2://username:password@host:port/dbname')
+
+        args = parser.parse_args()
+
+        return args
+
+    def get_command_value_from_name(self, command_name):
+        if command_name not in self._commandNames:
+            message = f'invalid choice: {command_name} (choose from {", ".join(self._commandNames)})'
+            raise argparse.ArgumentTypeError(message)
+
+        command_value = getattr(Commands, command_name, Commands.UNKNOWN)
+
+        return command_value

modules/Shared.py

@@ -0,0 +1,65 @@
+import logging
+import argparse
+
+
+class Constants:
+    APP_NAME = 'model-change-detector'
+    DATA_PIPELINE_EXECUTION_SCHEMA_NAME = 'data_pipeline'
+
+    class DataPipelineExecutionStatus:
+        STARTED = 'STARTED'
+        COMPLETED_SUCCESSFULLY = 'SUCCESSFUL'
+
+
+_logLevelStrings = [logging.getLevelName(logging.CRITICAL),
+                    logging.getLevelName(logging.ERROR),
+                    logging.getLevelName(logging.WARNING),
+                    logging.getLevelName(logging.INFO),
+                    logging.getLevelName(logging.DEBUG)]
+
+_defaultLogLevelString = logging.getLevelName(logging.INFO)
+
+
+def configure_root_logger(log_level):
+    # get the root logger
+    logger = logging.getLogger()
+
+    # with the given log level
+    logger.setLevel(logging.DEBUG)
+
+    # and one handler, at the same log level
+    console_stream_handler = logging.StreamHandler()
+    console_stream_handler.setLevel(log_level)
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    console_stream_handler.setFormatter(formatter)
+    logger.addHandler(console_stream_handler)
+
+    return
+
+
+def get_default_arguments():
+    parser = argparse.ArgumentParser(add_help=False)
+
+    parser.add_argument('-l', '--log-level',
+                        action='store',
+                        const=_defaultLogLevelString,
+                        default=_defaultLogLevelString,
+                        type=get_log_level_int_from_string,
+                        nargs='?',
+                        help=f'choose program\'s logging level, from {", ".join(_logLevelStrings)}; '
+                             f'default is {_defaultLogLevelString}')
+
+    return parser
+
+
+def get_log_level_int_from_string(log_level_string):
+    if log_level_string not in _logLevelStrings:
+        message = f'invalid choice: {log_level_string} (choose from {", ".join(_logLevelStrings)})'
+        raise argparse.ArgumentTypeError(message)
+
+    log_level_int = getattr(logging, log_level_string, logging.getLevelName(_defaultLogLevelString))
+
+    # check the logging log_level_choices have not changed from our expected values
+    assert isinstance(log_level_int, int)
+
+    return log_level_int

modules/commands/BaseCommand.py

@@ -0,0 +1,13 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from modules.DataPipelineExecutionRepository import DataPipelineExecutionRepository
+from modules.BaseObject import BaseObject
+
+
+class BaseCommand(BaseObject):
+    def __init__(self, db_connection_string, logger=None):
+        super().__init__(logger)
+        self.db_engine = create_engine(db_connection_string, echo=False)
+        self.session_maker = sessionmaker(bind=self.db_engine)
+        self.repository = DataPipelineExecutionRepository(self.session_maker)
+        self.repository.create_schema(engine=self.db_engine)

modules/commands/CommandFactory.py

@@ -0,0 +1,16 @@
+from modules.commands.StartCommand import StartCommand
+from modules.BaseObject import BaseObject
+
+
+class CommandFactory(BaseObject):
+    def __init__(self, logger=None):
+        super().__init__(logger)
+        self.commandTypes = [StartCommand]
+
+    def create_command(self, command, db_connection_string):
+        for commandType in self.commandTypes:
+            if commandType.can_execute_command(command):
+                self.logger.debug(f'Found command type {commandType} for command {command}.')
+                return commandType(db_connection_string)
+
+            raise RuntimeError(f'There are no command types that can handle the {command} command')