Merge pull request #3 from PageUpPeopleOrg/feature/OSC-986-finish-execution

feature/OSC-986 add support to finish a data pipeline execution

Author: ChintanRaval

README.md

@@ -2,23 +2,36 @@
 
 ## About
 
-A utility that detects changes in models.  
+A utility that detects changes in models.
 
 ## Usage
 
 ```commandline
-py mcd.py <command> <db-connection-string> [--help] [--log-level]
+py mcd.py [options] <command> [command-parameters]
 ```
 
-- `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`
+- `options` include:
+  - `--help | -h`: displays help menu.
+  - `--log-level | -l`: choose program's logging level, from CRITICAL, ERROR, WARNING, INFO, DEBUG; default is INFO.
+- `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. Returns an `execution-id` which is a GUID identifier of the new execution.
+    - `db-connection-string`: 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`
+  - `finish`: Marks the completion of an existing execution by updating a record for the same in the given database. Returns nothing unless there's an error.
+    - `db-connection-string`: 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`
+    - `execution-id`: a GUID identifier of an existing data pipeline execution.
+
+To get help,use:
+
+```
+py mcd.py --help
+py mcd.py <command> --help
+```
 
 ### 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]`
+- `py mcd.py [options] <command> [command-parameters]`
 
 _Windows example:_
 
@@ -28,7 +41,7 @@ new-env\scripts\activate
 
 py -m pip install -r requirements.txt
 
-py mcd.py START postgresql+psycopg2://user:password@host:port/dbname
+py mcd.py start postgresql+psycopg2://user:password@host:port/dbname
 ```
 
 ### As a package
@@ -38,7 +51,7 @@ py mcd.py START postgresql+psycopg2://user:password@host:port/dbname
 - [Install](https://pip.pypa.io/en/stable/reference/pip_install/#editable-installs) this package
   - `pip install -e path/to/ProjectX`
   - `pip install -e git+git://github.com/ProjectX.git#egg=ProjectX`
-- `py -m mcd <command> <db-connection-string> [--help] [--log-level]`
+- `py -m mcd [options] <command> <command-options>`
 
 _Windows example:_
 
@@ -51,15 +64,15 @@ new-env\scripts\activate
 
 pip install -e git+git://github.com/PageUpPeopleOrg/model-change-detector.git#egg=mcd
 
-py -m mcd START postgresql+psycopg2://user:password@host:port/dbname
+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 
+4. Develop and test code changes
 5. Once done, deactivate the virtual environment
 
 ### Install pre-requisites
@@ -96,13 +109,15 @@ On Windows:
 
 On Linux / Mac OS
 
- `source path/to/environment/bin/activate` _e.g._ `source new-env/bin/activate`
+`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.:
 
-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
@@ -113,7 +128,7 @@ pip install -r requirements.txt
 
 ### Deactivate the virtual environment
 
-Once done, deactivate the virtual environment with a simple `decativate` command, e.g.: 
+Once done, deactivate the virtual environment with a simple `decativate` command, e.g.:
 
 ```commandline
 (new-env) C:\path\to\working\dir: deactivate

mcd.py

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

modules/DataPipelineExecutionRepository.py

@@ -28,3 +28,13 @@ def get_last_successful_data_load_execution(self):
             .order_by(desc(DataPipelineExecutionEntity.last_updated_on))\
             .first()
 
+    def finish_existing(self, execution_id):
+        session = self.session_maker()
+        data_pipeline_execution = session.query(DataPipelineExecutionEntity)\
+            .filter_by(id=execution_id)\
+            .first()
+        data_pipeline_execution.status = Constants.DataPipelineExecutionStatus.COMPLETED_SUCCESSFULLY
+        session.commit()
+        return data_pipeline_execution
+
+

modules/ModelChangeDetector.py

@@ -2,52 +2,58 @@
 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
+from modules.commands.StartCommand import StartCommand
+from modules.commands.FinishCommand import FinishCommand
 
 
 class ModelChangeDetector(BaseObject):
-
-    _commandNames = [Commands.get_name(Commands.START)]
-
     def __init__(self, logger=None):
-        self.args = self.get_arguments()
+        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()
+        self.args.func()
+
+    def __process_start_command(self):
+        StartCommand(self.args.db_connection_string).execute()
 
-    def main(self):
-        command_executor = self.command_factory.create_command(self.args.command, self.args.db_connection_string)
-        command_executor.execute()
+    def __process_finish_command(self):
+        FinishCommand(self.args.db_connection_string, self.args.execution_id).execute()
 
-    def get_arguments(self):
+    def __get_arguments(self):
         parser = argparse.ArgumentParser(description=Constants.APP_NAME,
+                                         usage='mcd [options] <command> [command-parameters]\n\n'
+                                               'To see help text, you can run\n'
+                                               '  mcd --help\n'
+                                               '  mcd <command> --help',
                                          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..')
+        subparsers = parser.add_subparsers(title='commands', metavar='', dest='command')
 
-        parser.add_argument('db_connection_string',
-                            metavar='db-connection-string',
-                            help='provide in PostgreSQL & Psycopg format, '
-                                 'postgresql+psycopg2://username:password@host:port/dbname')
+        start_command_parser = subparsers.add_parser('start', help='help text for \'start\' command')
+        start_command_parser.set_defaults(func=self.__process_start_command)
+        self.__get_default_command_arguments(start_command_parser)
+
+        finish_command_parser = subparsers.add_parser('finish', help='help text for \'finish\' command')
+        finish_command_parser.set_defaults(func=self.__process_finish_command)
+        self.__get_default_command_arguments(finish_command_parser)
+        finish_command_parser.add_argument('execution_id',
+                                           metavar='execution_id',
+                                           help='data pipeline execution id as received using \'start\' command')
 
         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)
+    @staticmethod
+    def __get_default_command_arguments(command_parser):
+        command_parser.add_argument('db_connection_string',
+                                    metavar='db-connection-string',
+                                    help='provide in PostgreSQL & Psycopg format, '
+                                         'postgresql+psycopg2://username:password@host:port/dbname')
 
-        return command_value

modules/commands/CommandFactory.py

@@ -0,0 +1,11 @@
+from modules.commands.BaseCommand import BaseCommand
+
+
+class FinishCommand(BaseCommand):
+    def __init__(self, db_connection_string, execution_id, logger=None):
+        super().__init__(db_connection_string, logger)
+        self._executionId = execution_id
+
+    def execute(self):
+        data_pipeline_execution = self.repository.finish_existing(self._executionId)
+        self.logger.debug('Finised data_pipeline_execution = ' + str(data_pipeline_execution))

modules/commands/Commands.py

@@ -1,15 +1,10 @@
-from modules.commands import Commands
 from modules.commands.BaseCommand import BaseCommand
 
 
 class StartCommand(BaseCommand):
     def __init__(self, db_connection_string, logger=None):
         super().__init__(db_connection_string, logger)
 
-    @staticmethod
-    def can_execute_command(command):
-        return command == Commands.START
-
     def execute(self):
         data_pipeline_execution = self.repository.start_new()
         self.logger.debug('Started new data_pipeline_execution = ' + str(data_pipeline_execution))