Merge pull request #29 from pageuppeople-opensource/add-ability-to-accept-initial-execution-id

Add ability to accept initial execution id

Author: ChintanRaval

README.md

@@ -1,6 +1,6 @@
 # Data Pipeline Orchestrator
 
-[![Build Status](https://travis-ci.com/PageUpPeopleOrg/data-pipeline-orchestrator.svg?branch=master)](https://travis-ci.com/PageUpPeopleOrg/data-pipeline-orchestrator)
+[![Build Status](https://travis-ci.com/pageuppeople-opensource/data-pipeline-orchestrator.svg?branch=master)](https://travis-ci.com/pageuppeople-opensource/data-pipeline-orchestrator)
 
 A utility that persists state of a data pipeline execution and uses them to detect changes in models.
 
@@ -16,6 +16,7 @@ $ python -m dpo [options] <db-connection-string> <command> [command-args]
 - `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`
 - `command` is the function to be performed by the utility. The currently supported values are:
   - `init-execution`: Marks the start of a new execution. Returns an `execution-id` which is a GUID identifier of the new execution.
+    - `execution-id`: An optional GUID to use as the execution-id of new execution. Supports a PostgreSQL UUID type value. Throws an error if the GUID provided is already in use.
   - `get-last-successful-execution`: Finds the last successful execution. Returns an `execution-id` which is a GUID identifier of the new execution, if found; else returns and empty string.
   - `get-execution-completion-timestamp`: Returns the `last-updated-on` timestamp with timezone of the given `execution-id`. Raises error if given `execution-id` is invalid.
     - `execution-id`: a GUID identifier of an existing execution.

dpo/DataPipelineOrchestrator.py

@@ -26,7 +26,7 @@ def __init__(self, logger=None):
         self.args.func()
 
     def __process_init_execution_command(self):
-        InitialiseExecutionCommand(self.args.db_connection_string).execute()
+        InitialiseExecutionCommand(self.args.db_connection_string, self.args.execution_id).execute()
 
     def __process_get_last_successful_execution_command(self):
         GetLastSuccessfulExecutionCommand(self.args.db_connection_string).execute()
@@ -70,6 +70,16 @@ def __get_arguments(self):
         init_execution_command_parser = subparsers.add_parser(
             'init-execution', help='initialises a new execution')
         init_execution_command_parser.set_defaults(func=self.__process_init_execution_command)
+        init_execution_command_parser.add_argument(
+            '-id', '--execution-id',
+            action='store',
+            const=None,
+            default=None,
+            type=str,
+            nargs='?',
+            help='An optional GUID to use as the execution-id of new execution.'
+                 'Supports a PostgreSQL UUID type value.'
+                 'Throws an error if the GUID provided is already in use.')
 
         get_last_successful_execution_command_parser = subparsers.add_parser(
             'get-last-successful-execution',

dpo/DataRepository.py

@@ -16,10 +16,11 @@ def __init__(self, db_engine, logger=None):
     def get_current_db_datetime_with_timezone(self):
         return self.db_engine.execute(select([func.now()])).fetchone()[0]
 
-    def initialise_execution(self):
+    def initialise_execution(self, execution_id=None):
         session = self.session_maker()
 
         data_pipeline_execution = ExecutionEntity()
+        data_pipeline_execution.execution_id = execution_id if execution_id is not None else data_pipeline_execution.execution_id
         session.add(data_pipeline_execution)
 
         session.commit()

dpo/commands/InitialiseExecutionCommand.py

@@ -2,11 +2,12 @@
 
 
 class InitialiseExecutionCommand(BaseCommand):
-    def __init__(self, db_connection_string, logger=None):
+    def __init__(self, db_connection_string, execution_id=None, logger=None):
         super().__init__(db_connection_string, logger)
+        self._execution_id = execution_id
 
     def execute(self):
-        data_pipeline_execution = self.repository.initialise_execution()
+        data_pipeline_execution = self.repository.initialise_execution(execution_id=self._execution_id)
         self.logger.debug('Initialised new data_pipeline_execution = ' + str(data_pipeline_execution))
         self.output(data_pipeline_execution.execution_id)
 

setup.py

@@ -1,7 +1,7 @@
 from setuptools import setup, find_packages
 
 setup(name='dpo',
-      version='0.1.1',
+      version='0.1.6-beta',
       packages=find_packages(),
       install_requires=[
           'psycopg2-binary==2.8.2',

tests/integration/test_integration.sh

@@ -59,7 +59,13 @@ AssertAreEqual () {
 }
 
 InitialiseExecution () {
-    local executionId=$($dpo $dpo_conn_str init-execution)
+    local providedExecutionId=$1
+    if [ ! -z "$providedExecutionId" ]
+    then
+        local executionId=$($dpo $dpo_conn_str init-execution --execution-id "$providedExecutionId")
+    else
+        local executionId=$($dpo $dpo_conn_str init-execution)
+    fi
 
     if [ ${#executionId} != 36 ]
     then
@@ -135,15 +141,22 @@ CompleteExecution () {
 ExecuteAndAssert () {
     # Arrange
     local iter_no=$1
-    local loadRowsProcessed=$2
-    local transformRowsProcessed=$3
-    local expected_lastSuccessfulExecId=$4
-    local expected_changedLoadModels=$5
-    local expected_changedTransformModels=$6
-    local __resultvar=$7
+    local providedExecutionId=$2
+    local loadRowsProcessed=$3
+    local transformRowsProcessed=$4
+    local expected_lastSuccessfulExecId=$5
+    local expected_changedLoadModels=$6
+    local expected_changedTransformModels=$7
+    local __resultvar=$8
 
     # Act
-    local execId=$(InitialiseExecution)
+    if [ ! -z "$providedExecutionId" ]
+    then
+        local execId=$(InitialiseExecution $providedExecutionId)
+        AssertAreEqual "Iteration $iter_no's Initialised ExecutionID" "$execId" "$providedExecutionId"
+    else
+        local execId=$(InitialiseExecution)
+    fi
     echo "  iter$iter_no execId                                = $execId"
 
     local lastSuccessfulExecId=$(GetLastSuccessfulExecution)
@@ -192,6 +205,7 @@ transform_model_4="transform_model_4_$(NewUUID)"
 echo -e "\nBeginning execution #1"
 
 # ARRANGE
+iter1_execution_id=""
 ## Remove test models' directory, if exists
 rm -rf $modelDirectory
 
@@ -212,14 +226,15 @@ echo "transform_model_3" > "$transformModelDirectory/$transform_model_3.sql"
 iter1_expected_lastSuccessfulExecId="" # pass in empty string to skip test since we don't know the past state of the pipeline
 iter1_expected_changedLoadModels="$load_model_1 $load_model_2"
 iter1_expected_changedTransformModels="$transform_model_1 $transform_model_2 $transform_model_3"
-ExecuteAndAssert "1" "2147483647" "" "$iter1_expected_lastSuccessfulExecId" "$iter1_expected_changedLoadModels" "$iter1_expected_changedTransformModels" iter1_execId
+ExecuteAndAssert "1" "$iter1_execution_id" "2147483647" "" "$iter1_expected_lastSuccessfulExecId" "$iter1_expected_changedLoadModels" "$iter1_expected_changedTransformModels" iter1_execId
 
 ###############
 # Execution 2 #
 ###############
 echo -e "\nBeginning execution #2"
 
 # ARRANGE
+iter2_execution_id=$(NewUUID)
 echo " Making no changes to any LOAD models"
 
 echo " Modifying transform_model_2"
@@ -235,4 +250,4 @@ echo "transform_model_4" > "$transformModelDirectory/$transform_model_4.sql"
 iter2_expected_lastSuccessfulExecId="$iter1_execId"
 iter2_expected_changedLoadModels=""
 iter2_expected_changedTransformModels="$transform_model_2 $transform_model_4"
-ExecuteAndAssert "2" "9223372036854775807" "" "$iter2_expected_lastSuccessfulExecId" "$iter2_expected_changedLoadModels" "$iter2_expected_changedTransformModels" iter2_execId
+ExecuteAndAssert "2" "$iter2_execution_id" "9223372036854775807" "" "$iter2_expected_lastSuccessfulExecId" "$iter2_expected_changedLoadModels" "$iter2_expected_changedTransformModels" iter2_execId