Merge branches 'master' and 'process_tracker_python-142' of https://github.com/OpenDataAlex/process_tracker_python into process_tracker_python-142

Author: Alex Meadows

.travis.yml

@@ -49,9 +49,7 @@ jobs:
       user: "__token__"
       distributions: sdist bdist_wheel
       password:
-#        secure: jYe8ygGYuJox5f1vQvraIDNlR2Yk/+/afIXdsxaxRi/QN135rIHxRih24iYv8ZQewfW+w4CveEPz2pr0ew5NG8hm2MD5hQZI/A2QE4BC53Tr3xMAWSdPQgcI1rX4Ieu6BHTS3jet212yLuBk41GdASKztz+s/CVvVqT0A29U1b85OVzoVHz6OTgaM1m0pKeJE6Ho66r5O6J7ClZfEY4RJ8jROmrT7ufW3hEs/VQXwfAUWcjy6l5H5VYTLa/g94MgQ+0WmgcdUbB8s3c2q7n4RiaVEowZOLtBmRElDIPiLJC8JGAIFKcf5W8pzXK/ZhZhY3ppgmBvRVeyPdiAwjysionzl4vIlNxC28acI3HzvJbAdvwPsuMD+pnTo/LKRqqvGey2NSnMPmzGXxzqVUqQ5Wa1Yzdcqg8mbWHFRw9icMWKM84CnDBvqkmJcfnj5R4X6Rc77mr9jc5LMMBBblNtY7dQc2AooframujGA2mP1xABQ9vOSg89MdnlhHrgEK6m+Pz0S2hhlsGRw9yMF9L4Q1ie7xKeE49L3HKZSjKEgSMv9y9RbqnmW32F0UoKgWkCJE8DV9wbZe5Htn+lpwmXI0vu+cxnqsiZvG8wjm+klJGn5F84OFTN3zQOC/kIBXQXvutZYtgF0vofdCmKQzK9istl16Eur1r4NCbov3oM97o=
         secure: G0hpSH9oaQIQ4BA06R8ygAAG+2urF7bllSYDdqgt6JdyseczhpCIrDI4CvL8Nj8jR4KqqB7Wm4KcEZrq48VGdYCSAt/dawtrFmSoDDlyTMxpudl8sZPczyDlTPCzH3ey2aFhSlT/m1kXMimvl6HunBc0Jm9l6Q29VwZx1KvOcyTB/HU9t1ucCmW8uVXMZMgRymTFRPLCl7R5wKfD9aTvKt71k8MceMHBAV6pqILP/QcNUiMMiM78Pe5J5i2auD7wB06RSkAUkqbsX45PaCkO2X35OQ7yTUfry9xn2J0ouSFf4bAAq/sp+SFudLEh7qshGsUkXkVgOd4j9GHhltPpT6EzjaTa5E6uDScqovW0Q5XIQFE+lOWHYBw89vzaycZ5n6n8z8J1U6fsXVLbUfjbr+R3Nq9ncKCY36ygoPdQi/hDdAo/N+gTCbOB9mN6t78KPcg4qu4OoblfeRbvwW2r4caaHXDpqlY+OkRth9pz3GJV9gprhNlEISRbznMZ2EE+ZGdh1CU+TeH8ftLZfM/qQTlIGjSxHBNmlR4En8or4za1iIdqKheScgBtRgkV29hN1/93hniK5z5PRsQte3fhv58gPXrYQI8ymGP2ZSrPnXZJifOdr9Lu1/o8GcbpzlHz9NjD2mLrErSJbqM9eYPyR14XVfYVwNLLGd72iwzW7y8=
-
       on:
         tags: true
       skip_cleanup: true

dbscripts/mysql_process_tracker.sql

@@ -74,7 +74,7 @@ create table extract_filetype_lkup
 (
 	extract_filetype_id int auto_increment
 		primary key,
-	extract_filetype_code varchar(5) not null,
+	extract_filetype_code varchar(25) not null,
 	extract_filetype varchar(75) not null,
 	delimiter_char char null,
 	quote_char char null,
@@ -338,6 +338,8 @@ create table process_tracking
 	process_run_actor_id int null,
 	is_latest_run tinyint(1) not null,
 	process_run_name varchar(250) null,
+	process_run_insert_count int default 0 not null,
+	process_run_update_count int default 0 not null,
 	constraint process_tracking_process_run_name_uindex
 		unique (process_run_name),
 	constraint process_tracking_ibfk_1

dbscripts/mysql_process_tracker_defaults.sql

@@ -20,7 +20,7 @@ INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VAL
 INSERT INTO process_tracker.process_type_lkup (process_type_id, process_type_name) VALUES (default, 'Extract');
 INSERT INTO process_tracker.process_type_lkup (process_type_id, process_type_name) VALUES (default, 'Load');
 
-INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VALUES (default, 'version', '0.8.0');
+INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VALUES (default, 'version', '0.8.1');
 
 INSERT INTO process_tracker.extract_compression_type_lkup (extract_compression_type_id, extract_compression_type) VALUES (1, 'zip');
 

dbscripts/postgresql_process_tracker.sql

@@ -251,7 +251,7 @@ create table extract_filetype_lkup
 	extract_filetype_id serial not null
 		constraint extract_filetype_lkup_pk
 			primary key,
-	extract_filetype_code varchar(5) not null,
+	extract_filetype_code varchar(25) not null,
 	extract_filetype varchar(75) not null,
 	delimiter_char char,
 	quote_char char,
@@ -365,7 +365,9 @@ create table process_tracking
 		constraint process_tracking_fk03
 			references actor_lkup,
 	is_latest_run boolean default false,
-	process_run_name varchar(250) null
+	process_run_name varchar(250) null,
+	process_run_insert_count int default 0 not null,
+	process_run_update_count int default 0 not null
 );
 
 comment on table process_tracking is 'Tracking table of process runs.';

dbscripts/postgresql_process_tracker_defaults.sql

@@ -21,7 +21,7 @@ INSERT INTO process_tracker.error_type_lkup (error_type_id, error_type_name) VAL
 INSERT INTO process_tracker.process_type_lkup (process_type_id, process_type_name) VALUES (default, 'Extract');
 INSERT INTO process_tracker.process_type_lkup (process_type_id, process_type_name) VALUES (default, 'Load');
 
-INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VALUES (default, 'version', '0.8.0');
+INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VALUES (default, 'version', '0.8.1');
 
 INSERT INTO process_tracker.extract_compression_type_lkup (extract_compression_type_id, extract_compression_type) VALUES (1, 'zip');
 

process_tracker/__init__.py

@@ -1,4 +1,4 @@
 from process_tracker.extract_tracker import ExtractTracker
 from process_tracker.process_tracker import ProcessTracker
 
-__version__ = "0.8.0"
+__version__ = "0.8.1"

process_tracker/models/extract.py

@@ -74,7 +74,7 @@ class ExtractFileType(Base, BaseColumn):
         primary_key=True,
         nullable=False,
     )
-    extract_filetype_code = Column(String(5), nullable=False)
+    extract_filetype_code = Column(String(25), nullable=False)
     extract_filetype = Column(String(75), nullable=False, unique=True)
     delimiter_char = Column(String(1), nullable=True)
     quote_char = Column(String(1), nullable=True)

process_tracker/models/process.py

@@ -508,6 +508,8 @@ class ProcessTracking(Base, BaseColumn):
     process_run_start_date_time = Column(DateTime, nullable=False)
     process_run_end_date_time = Column(DateTime, nullable=True)
     process_run_record_count = Column(Integer, nullable=False, default=0)
+    process_run_insert_count = Column(Integer, nullable=False, default=0)
+    process_run_update_count = Column(Integer, nullable=False, default=0)
     process_run_actor_id = Column(
         Integer, ForeignKey("process_tracker.actor_lkup.actor_id"), nullable=False
     )

process_tracker/process_tracker.py

@@ -73,12 +73,13 @@ def __init__(
         config_location=None,
         dataset_types=None,
         schedule_frequency=None,
-        process_run_id=None,
+        process_tracking_id=None,
     ):
         """
         ProcessTracker is the primary engine for tracking data integration processes.
         :param process_name: Name of the process being tracked.
         :param process_run_name: Optional name of the process run.
+        :param process_type: Type of process the process_name is.  Optional if process already exists.
         :param actor_name: Name of the person or environment runnning the process.
         :param tool_name: Name of the tool used to run the process.
         :param sources: A single source name or list of source names for the given process. If source_objects is set,
@@ -104,9 +105,9 @@ def __init__(
         :type dataset_types: list
         :param schedule_frequency: The general scheduling frequency for the process (i.e. hourly)
         :type schedule_frequency: string
-        :param process_run_id: If trying to access an already running process, provide the process run's id.
+        :param process_tracking_id: If trying to access an already running process, provide the process run's id.
         Object will be built for that specific process run.
-        :type process_run_id: int
+        :type process_tracking_id: int
         """
         self.config_location = config_location
         self.config = SettingsManager(config_location=self.config_location)
@@ -128,11 +129,13 @@ def __init__(
         self.process_status_failed = self.process_status_types["failed"]
         self.process_status_hold = self.process_status_types["on hold"]
 
-        if process_run_id is not None:
+        if process_tracking_id is not None:
             self.logger.info("Process run id provided.  Checking if exists.")
 
             process_run = self.data_store.get_or_create_item(
-                model=ProcessTracking, process_tracking_id=process_run_id, create=False
+                model=ProcessTracking,
+                process_tracking_id=process_tracking_id,
+                create=False,
             )
 
             if process_run is not None:
@@ -145,32 +148,32 @@ def __init__(
 
                 self.dataset_types = process_run.process.dataset_types
                 self.sources = self.determine_process_sources(
-                    process_run_id=process_run_id
+                    process_run_id=process_tracking_id
                 )
                 self.targets = self.determine_process_targets(
-                    process_run_id=process_run_id
+                    process_run_id=process_tracking_id
                 )
 
                 self.process_name = process_run.process.process_name
                 self.process_tracking_run = process_run
                 self.process_run_name = process_run.process_run_name
 
             else:
-                error_msg = "Process run not found based on id %s." % process_run_id
+                error_msg = (
+                    "Process run not found based on id %s." % process_tracking_id
+                )
                 self.logger.error(error_msg)
                 raise Exception(error_msg)
         else:
-            if process_name is None or process_type is None:
-                error_msg = "process_name and process_type must be set."
+            if process_name is None is None:
+                error_msg = "process_name must be set."
                 self.logger.error(error_msg)
                 raise Exception(error_msg)
 
             self.actor = self.data_store.get_or_create_item(
                 model=Actor, actor_name=actor_name
             )
-            self.process_type = self.data_store.get_or_create_item(
-                model=ProcessType, process_type_name=process_type
-            )
+
             self.tool = self.data_store.get_or_create_item(
                 model=Tool, tool_name=tool_name
             )
@@ -184,13 +187,27 @@ def __init__(
                     model=ScheduleFrequency, schedule_frequency_name=schedule_frequency
                 )
 
-            self.process = self.data_store.get_or_create_item(
-                model=Process,
-                process_name=process_name,
-                process_type_id=self.process_type.process_type_id,
-                process_tool_id=self.tool.tool_id,
-                schedule_frequency_id=self.schedule_frequency.schedule_frequency_id,
-            )
+            if process_type is None:
+
+                self.process = self.data_store.get_or_create_item(
+                    model=Process, process_name=process_name, create=False
+                )
+
+                self.process_type = self.process.process_type
+
+            else:
+
+                self.process_type = self.data_store.get_or_create_item(
+                    model=ProcessType, process_type_name=process_type
+                )
+
+                self.process = self.data_store.get_or_create_item(
+                    model=Process,
+                    process_name=process_name,
+                    process_type_id=self.process_type.process_type_id,
+                    process_tool_id=self.tool.tool_id,
+                    schedule_frequency_id=self.schedule_frequency.schedule_frequency_id,
+                )
 
             # Dataset types should be loaded before source and target because they are also used there.
 
@@ -325,7 +342,7 @@ def determine_hold_status(self, last_run_status, last_run_id):
         Based on the setting 'max_concurrent_failures', count the number of failures for that number of process runs.
         If the counts match, process will remain on hold.  If last run is 'on_hold' process will remain on hold.
         :param last_run_status: The status of the previous run
-        :param last_run_id:  The process_run_id of the previous run
+        :param last_run_id:  The process_tracking_id of the previous run
         :return:
         """
         self.logger.debug("Determining if process should be put on or remain on hold.")
@@ -362,7 +379,7 @@ def determine_hold_status(self, last_run_status, last_run_id):
 
     def determine_process_sources(self, process_run_id):
         """
-        Based on the process_run_id, find the given process' sources - either at the attribute, object, or source level
+        Based on the process_tracking_id, find the given process' sources - either at the attribute, object, or source level
         :param process_run_id: Process run identifier
         :type process_run_id: int
         :return: Array of source objects at lowest granularity.
@@ -425,7 +442,7 @@ def determine_process_sources(self, process_run_id):
 
     def determine_process_targets(self, process_run_id):
         """
-        Based on the process_run_id, find the given process' targets - either at the attribute, object, or source level
+        Based on the process_tracking_id, find the given process' targets - either at the attribute, object, or source level
         :param process_run_id: Process run identifier
         :type process_run_id: int
         :return: Array of source objects used as target for the process at lowest granularity.
@@ -641,13 +658,13 @@ def find_process_by_schedule_frequency(self, frequency="daily"):
         process_list = list()
 
         processes = (
-            self.data_store.session.query(Process.process_id)
+            self.data_store.session.query(Process)
             .join(ScheduleFrequency)
             .filter(ScheduleFrequency.schedule_frequency_name == frequency)
         )
 
         for process in processes:
-            process_list.append(process.process_id)
+            process_list.append(process)
 
         return process_list
 
@@ -1319,22 +1336,53 @@ def set_process_run_low_high_dates(self, low_date=None, high_date=None):
 
         self.session.commit()
 
-    def set_process_run_record_count(self, num_records):
+    def record_count_manager(self, original_count, num_records):
+        """
+        Given two record counts, one the overall count and the other the current count, process and return the adjusted
+        amount.
+        :param original_count: The original overall count of records
+        :type original_count: int
+        :param num_records: The adjusted count of records
+        :type num_records: int
+        """
+        if original_count == 0 or original_count is None:
+            return_count = num_records
+        else:
+            return_count = original_count + (num_records - original_count)
+
+        return return_count
+
+    def set_process_run_record_count(self, num_records, processing_type=None):
         """
         For the given process run, set the process_run_record_count for the number of records processed.  Will also
         update the process' total_record_count - the total number of records ever processed by that process
-        :param num_records:
+        :param num_records: Count of number of records processed.
+        :type num_records: int
+        :param processing_type: Type of records being processed.  Valid values: None, insert, update.  None will only
+                                update overall counts
         :return:
         """
         process_run_records = self.process.total_record_count
+        process_run_inserts = self.process_tracking_run.process_run_insert_count
+        process_run_updates = self.process_tracking_run.process_run_update_count
 
-        if process_run_records == 0 or process_run_records is None:
-
-            self.process.total_record_count = num_records
-        else:
-            self.process.total_record_count = self.process.total_record_count + (
-                num_records - process_run_records
+        if processing_type == "insert":
+            self.process_tracking_run.process_run_insert_count = self.record_count_manager(
+                original_count=process_run_inserts, num_records=num_records
+            )
+        elif processing_type == "update":
+            self.process_tracking_run.process_run_update_count = self.record_count_manager(
+                original_count=process_run_updates, num_records=num_records
             )
+        elif processing_type is not None:
+            error_msg = "Processing type not recognized."
+            self.logger.error(error_msg)
+            raise Exception(error_msg)
+
+        self.process.total_record_count = self.record_count_manager(
+            original_count=process_run_records, num_records=num_records
+        )
 
         self.process_tracking_run.process_run_record_count = num_records
+
         self.session.commit()

process_tracker/utilities/settings.py

@@ -40,10 +40,7 @@ def __init__(self, config_location=None):
         else:
             self.config_path = config_location
 
-            if (
-                "process_tracker_config.ini" not in self.config_path
-                and ".ini" not in self.config_path
-            ):
+            if ".ini" not in self.config_path:
                 self.logger.debug(
                     "process_tracker_config.ini not present.  Appending to %s"
                     % self.config_path
@@ -59,7 +56,7 @@ def __init__(self, config_location=None):
                 self.logger.debug("Config file is now %s" % self.config_file)
             else:
                 self.logger.debug(
-                    "process_tracker_config.ini present.  Setting config_path to config_file."
+                    "process_tracker config file present in path.  Setting config_path to config_file."
                 )
                 self.config_file = self.config_path