Merge pull request #36 from OpenDataAlex/process_tracker_python-17

Process tracker python 17

Author: OpenDataAlex

dbscripts/mysql_process_tracker.sql

@@ -51,30 +51,33 @@ create table location_lkup
 		foreign key (location_type) references location_type_lkup (location_type_id)
 );
 
-create table extract_tracking
+create table process_tracker.extract_tracking
 (
 	extract_id int auto_increment
 		primary key,
 	extract_filename varchar(750) not null,
 	extract_location_id int null,
 	extract_status_id int null,
 	extract_registration_date_time datetime not null,
+	extract_write_low_date_time datetime null comment 'The lowest datetime of the data set as noted when writing the data file.',
+	extract_write_high_date_time datetime null comment 'The highest datetime of the data set as noted when writing the data file.',
+	extract_write_record_count int null comment 'The record count of the data set as noted when writing the data file.',
+	extract_load_low_date_time datetime null comment 'The lowest datetime of the data set as noted when loading the data file.  Should match the extract_write_low_date_time.',
+	extract_load_high_date_time datetime null comment 'The highest datetime of the data set as noted when loading the data file. Should match the extract_load_high_date_time.',
+	extract_load_record_count int null comment 'The record count of the data set when loading the data file.',
 	constraint extract_filename
 		unique (extract_filename),
 	constraint extract_tracking_ibfk_1
-		foreign key (extract_location_id) references location_lkup (location_id),
+		foreign key (extract_location_id) references process_tracker.location_lkup (location_id),
 	constraint extract_tracking_ibfk_2
-		foreign key (extract_status_id) references extract_status_lkup (extract_status_id)
+		foreign key (extract_status_id) references process_tracker.extract_status_lkup (extract_status_id)
 );
 
 create index extract_location_id
-	on extract_tracking (extract_location_id);
+	on process_tracker.extract_tracking (extract_location_id);
 
 create index extract_status_id
-	on extract_tracking (extract_status_id);
-
-create index location_type
-	on location_lkup (location_type);
+	on process_tracker.extract_tracking (extract_status_id);
 
 create table process_tracker.extract_dependency
 (

dbscripts/postgresql_process_tracker.sql

@@ -298,7 +298,13 @@ create table extract_tracking
 	extract_status_id integer
 		constraint extract_tracking_fk02
 			references extract_status_lkup,
-	extract_registration_date_time timestamp not null
+	extract_registration_date_time timestamp not null,
+	extract_write_low_date_time timestamp,
+	extract_write_high_date_time timestamp,
+	extract_write_record_count integer,
+	extract_load_low_date_time timestamp,
+	extract_load_high_date_time timestamp,
+	extract_load_record_count integer
 );
 
 comment on table extract_tracking is 'Tracking table for all extract/staging data files.';
@@ -313,6 +319,18 @@ comment on column extract_tracking.extract_status_id is 'The status of the extra
 
 comment on column extract_tracking.extract_registration_date_time is 'The datetime that the extract was loaded into extract tracking.';
 
+comment on column extract_tracking.extract_write_low_date_time is 'The lowest datetime of the data set as noted when writing the data file.';
+
+comment on column extract_tracking.extract_write_high_date_time is 'The highest datetime of the data set as noted when writing the data file.';
+
+comment on column extract_tracking.extract_write_record_count is 'The record count of the data set as noted when writing the data file.';
+
+comment on column extract_tracking.extract_load_low_date_time is 'The lowest datetime of the data set as noted when loading the data file.  Should match the extract_write_low_date_time.';
+
+comment on column extract_tracking.extract_load_high_date_time is 'The highest datetime of the data set as noted when loading the data file.';
+
+comment on column extract_tracking.extract_load_record_count is 'The record count of the data set when loading the data file.';
+
 alter table extract_tracking owner to pt_admin;
 
 create table extract_process_tracking

process_tracker/extract_tracker.py

@@ -9,6 +9,7 @@
 from process_tracker.data_store import DataStore
 from process_tracker.location_tracker import LocationTracker
 from process_tracker.utilities.settings import SettingsManager
+from process_tracker.utilities import utilities
 from process_tracker.models.extract import (
     Extract,
     ExtractDependency,
@@ -252,3 +253,75 @@ def retrieve_extract_process(self):
             self.session.commit()
 
         return extract_process
+
+    def set_extract_low_high_dates(self, low_date, high_date, audit_type="load"):
+        """
+        For the given extract, find the low and high date_times while writing or loading.
+        :param low_date: The low date of the data set.
+        :type low_date: Datetime/timestamp
+        :param high_date: The high date of the data set.
+        :type high_date: Datetime/timestamp.
+        :param audit_type: The type of audit fields being populated.  Valid types:  write, load
+        :type audittype: String
+        :return:
+        """
+
+        if audit_type == "write":
+
+            previous_low_date_time = self.extract.extract_write_low_date_time
+            previous_high_date_time = self.extract.extract_write_high_date_time
+
+            if utilities.determine_low_high_date(
+                date=low_date, previous_date=previous_low_date_time, date_type="low"
+            ):
+                self.extract.extract_write_low_date_time = low_date
+
+            if utilities.determine_low_high_date(
+                date=high_date, previous_date=previous_high_date_time, date_type="high"
+            ):
+                self.extract.extract_write_high_date_time = high_date
+
+        elif audit_type == "load":
+
+            previous_low_date_time = self.extract.extract_load_low_date_time
+            previous_high_date_time = self.extract.extract_load_high_date_time
+
+            if utilities.determine_low_high_date(
+                date=low_date, previous_date=previous_low_date_time, date_type="low"
+            ):
+                self.extract.extract_load_low_date_time = low_date
+
+            if utilities.determine_low_high_date(
+                date=high_date, previous_date=previous_high_date_time, date_type="high"
+            ):
+                self.extract.extract_load_high_date_time = high_date
+
+        else:
+            self.logger.error("%s is not a valid audit_type." % audit_type)
+            raise Exception("%s is not a valid audit_type." % audit_type)
+
+        self.session.commit()
+
+    def set_extract_record_count(self, num_records, audit_type="load"):
+        """
+        For the given audit type, set the number of records for the given extract.
+        :param num_records: Number of records tracked in extract
+        :type num_records: int
+        :param audit_type: The type of audit being populated.  Valid types:  write, load.
+        :type audit_type: str
+        :return:
+        """
+
+        if audit_type == "write":
+
+            self.extract.extract_write_record_count = num_records
+
+        elif audit_type == "load":
+
+            self.extract.extract_load_record_count = num_records
+
+        else:
+            self.logger.error("%s is not a valid audit_type." % audit_type)
+            raise Exception("%s is not a valid audit_type." % audit_type)
+
+        self.session.commit()

process_tracker/models/extract.py

@@ -44,6 +44,12 @@ class Extract(Base):
     extract_registration_date_time = Column(
         DateTime, nullable=False, default=datetime.now()
     )
+    extract_write_low_date_time = Column(DateTime, nullable=True)
+    extract_write_high_date_time = Column(DateTime, nullable=True)
+    extract_write_record_count = Column(Integer, nullable=True)
+    extract_load_low_date_time = Column(DateTime, nullable=True)
+    extract_load_high_date_time = Column(DateTime, nullable=True)
+    extract_load_record_count = Column(Integer, nullable=True)
 
     extract_process = relationship("ExtractProcess", back_populates="process_extracts")
     locations = relationship("Location", foreign_keys=[extract_location_id])

process_tracker/process_tracker.py

@@ -12,6 +12,7 @@
 from process_tracker.location_tracker import LocationTracker
 from process_tracker.utilities.logging import console
 from process_tracker.utilities.settings import SettingsManager
+from process_tracker.utilities import utilities
 
 from process_tracker.models.actor import Actor
 from process_tracker.models.extract import (
@@ -446,14 +447,16 @@ def set_process_run_low_high_dates(self, low_date=None, high_date=None):
         previous_low_date_time = self.process_tracking_run.process_run_low_date_time
         previous_high_date_time = self.process_tracking_run.process_run_low_date_time
 
-        if low_date is not None and (
-            previous_low_date_time is None or low_date < previous_low_date_time
+        if utilities.determine_low_high_date(
+            date=low_date, previous_date=previous_low_date_time, date_type="low"
         ):
+
             self.process_tracking_run.process_run_low_date_time = low_date
 
-        if high_date is not None and (
-            previous_high_date_time is None or high_date > previous_high_date_time
+        if utilities.determine_low_high_date(
+            date=high_date, previous_date=previous_high_date_time, date_type="high"
         ):
+
             self.process_tracking_run.process_run_high_date_time = high_date
 
         self.session.commit()

process_tracker/utilities/settings.py

@@ -23,8 +23,9 @@ def __init__(self, config_location=None):
             )
 
         else:
-            self.config_path = os.path.join(
-                config_location, "process_tracker_config.ini"
+            self.config_path = config_location
+            self.config_file = os.path.join(
+                self.config_path, "process_tracker_config.ini"
             )
 
         exists = os.path.isfile(self.config_file)

process_tracker/utilities/utilities.py

@@ -0,0 +1,58 @@
+"""
+Space for generalized helpers that can be utilized across the entire framework.
+"""
+import logging
+
+from process_tracker.utilities.settings import SettingsManager
+
+config = SettingsManager().config
+
+logger = logging.getLogger(__name__)
+logger.setLevel(config["DEFAULT"]["log_level"])
+
+
+def determine_low_high_date(date, previous_date, date_type):
+    """
+    For the given dates and date type, determine if the date replaces the previous date or not.
+    :param date: The new datetime.
+    :type date: Datetime/timestamp
+    :param previous_date: The previous datetime that date is being compared to.
+    :type previous_date: Datetime/timestamp
+    :param date_type: Is the comparison for a low date or high date?  Valid values:  low, high
+    :type date_type: str
+    :return: Boolean if date replaces previous_date
+    """
+
+    if date_type == "low":
+
+        if date is not None and (previous_date is None or date < previous_date):
+            return True
+        else:
+            return False
+
+    elif date_type == "high":
+
+        if date is not None and (previous_date is None or previous_date > date):
+            return True
+        else:
+            return False
+
+    else:
+        logger.error("%s is not a valid date_type." % date_type)
+        raise Exception("%s is not a valid date_type." % date_type)
+
+
+def timestamp_converter(data_store_type, timestamp):
+    """
+    Helper function for when testing with data stores that have funky formats for stock dates with SQLAlchemy.
+    :param data_store_type: The type of data store
+    :param timestamp: The timestamp to be created.
+    :return:
+    """
+
+    if data_store_type == "mysql":
+        timestamp = timestamp.replace(microsecond=0)
+    else:
+        timestamp = timestamp
+
+    return timestamp