process_tracker_python-89 Add Process Schedule Frequency

✨ General scheduling for processes is now available

High level scheduling frequencies are now available to schedule processes
by.  This is no more fine grained that saying 'these jobs run hourly or monthly'.
Actual scheduling and execution still has to be performed.

Author: Alex Meadows

dbscripts/mysql_process_tracker.sql

@@ -1,5 +1,14 @@
 USE process_tracker;
 
+create table schedule_frequency_lkup
+(
+	schedule_frequency_id int auto_increment
+		primary key,
+	schedule_frequency_name varchar(25) not null,
+	constraint schedule_frequency_lkup_schedule_frequency_name_uindex
+		unique (schedule_frequency_name)
+);
+
 create table data_type_lkup
 (
 	data_type_id int auto_increment
@@ -210,14 +219,23 @@ create table process
 	process_type_id int null,
 	process_tool_id int null,
 	last_failed_run_date_time datetime not null,
+	schedule_frequency_id int default 0 not null,
 	constraint process_name
 		unique (process_name),
+	constraint process_fk03
+		foreign key (schedule_frequency_id) references schedule_frequency_lkup (schedule_frequency_id),
 	constraint process_ibfk_1
 		foreign key (process_type_id) references process_type_lkup (process_type_id),
 	constraint process_ibfk_2
 		foreign key (process_tool_id) references tool_lkup (tool_id)
 );
 
+create index process_tool_id
+	on process_tracker.process (process_tool_id);
+
+create index process_type_id
+	on process_tracker.process (process_type_id);
+
 create index process_tool_id
 	on process (process_tool_id);
 

dbscripts/mysql_process_tracker_defaults.sql

@@ -24,4 +24,12 @@ INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VA
 
 INSERT INTO process_tracker.extract_compression_type_lkup (extract_compression_type_id, extract_compression_type) VALUES (1, 'zip');
 
-INSERT INTO process_tracker.extract_filetype_lkup (extract_filetype_id, extract_filetype_code, extract_filetype, delimiter_char, quote_char, escape_char) VALUES (1, 'csv', 'Comma Separated Values', ',', '"', '/');
+INSERT INTO process_tracker.extract_filetype_lkup (extract_filetype_id, extract_filetype_code, extract_filetype, delimiter_char, quote_char, escape_char) VALUES (1, 'csv', 'Comma Separated Values', ',', '"', '/');
+
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (7, 'annually');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (3, 'daily');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (2, 'hourly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (5, 'monthly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (6, 'quarterly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (0, 'unscheduled');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (4, 'weekly');

dbscripts/postgresql_process_tracker.sql

@@ -4,6 +4,19 @@ create schema process_tracker;
 
 alter schema process_tracker owner to pt_admin;
 
+create table schedule_frequency_lkup
+(
+	schedule_frequency_id serial not null
+		constraint schedule_frequency_lkup_pk
+			primary key,
+	schedule_frequency_name varchar(25) not null
+);
+
+alter table schedule_frequency_lkup owner to pt_admin;
+
+create unique index schedule_frequency_lkup_schedule_frequency_name_uindex
+	on schedule_frequency_lkup (schedule_frequency_name);
+
 create table data_type_lkup
 (
 	data_type_id serial not null
@@ -180,38 +193,41 @@ alter table process_type_lkup owner to pt_admin;
 create unique index process_type_lkup_udx01
 	on process_type_lkup (process_type_name);
 
-create table process
+create table process_tracker.process
 (
 	process_id serial not null
 		constraint process_pk
 			primary key,
 	process_name varchar(250) not null,
 	total_record_count integer default 0 not null,
-	process_type_id integer null
+	process_type_id integer
 		constraint process_fk02
-			references process_type_lkup,
-	process_tool_id integer null
+			references process_tracker.process_type_lkup,
+	process_tool_id integer
 		constraint process_fk03
-			references tool_lkup,
-	last_failed_run_date_time timestamp default '1900-01-01 00:00:00'::timestamp without time zone not null
+			references process_tracker.tool_lkup,
+	last_failed_run_date_time timestamp default '1900-01-01 00:00:00'::timestamp without time zone not null,
+	schedule_frequency_id integer default 0 not null
+		constraint process_fk04
+			references process_tracker.schedule_frequency_lkup
 );
 
-comment on table process is 'Processes being tracked';
+comment on table process_tracker.process is 'Processes being tracked';
 
-comment on column process.process_name is 'Unique name for process.';
+comment on column process_tracker.process.process_name is 'Unique name for process.';
 
-comment on column process.total_record_count is 'Total number of records processed over all runs of process.';
+comment on column process_tracker.process.total_record_count is 'Total number of records processed over all runs of process.';
 
-comment on column process.process_type_id is 'The type of process being tracked.';
+comment on column process_tracker.process.process_type_id is 'The type of process being tracked.';
 
-comment on column process.process_tool_id is 'The type of tool used to execute the process.';
+comment on column process_tracker.process.process_tool_id is 'The type of tool used to execute the process.';
 
-comment on column process.last_failed_run_date_time is 'The last time the process failed to run.';
+comment on column process_tracker.process.last_failed_run_date_time is 'The last time the process failed to run.';
 
-alter table process owner to pt_admin;
+alter table process_tracker.process owner to pt_admin;
 
 create unique index process_udx01
-	on process (process_name);
+	on process_tracker.process (process_name);
 
 create table process_dependency
 (

dbscripts/postgresql_process_tracker_defaults.sql

@@ -25,4 +25,12 @@ INSERT INTO process_tracker.system_lkup (system_id, system_key, system_value) VA
 
 INSERT INTO process_tracker.extract_compression_type_lkup (extract_compression_type_id, extract_compression_type) VALUES (1, 'zip');
 
-INSERT INTO process_tracker.extract_filetype_lkup (extract_filetype_id, extract_filetype_code, extract_filetype, delimiter_char, quote_char, escape_char) VALUES (1, 'csv', 'Comma Separated Values', ',', '"', '/');
+INSERT INTO process_tracker.extract_filetype_lkup (extract_filetype_id, extract_filetype_code, extract_filetype, delimiter_char, quote_char, escape_char) VALUES (1, 'csv', 'Comma Separated Values', ',', '"', '/');
+
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (0, 'unscheduled');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (1, 'hourly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (2, 'daily');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (3, 'weekly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (4, 'monthly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (5, 'quarterly');
+INSERT INTO process_tracker.schedule_frequency_lkup (schedule_frequency_id, schedule_frequency_name) VALUES (6, 'annually');

process_tracker/models/process.py

@@ -147,6 +147,12 @@ class Process(Base):
     last_failed_run_date_time = Column(
         DateTime(timezone=True), nullable=False, default=default_date
     )
+    schedule_frequency_id = Column(
+        Integer,
+        ForeignKey("process_tracker.schedule_frequency_lkup.schedule_frequency_id"),
+        nullable=False,
+        default=0,
+    )
 
     cluster_processes = relationship("ClusterProcess", passive_deletes="all")
     process_tracking = relationship("ProcessTracking", passive_deletes="all")

process_tracker/models/schedule.py

@@ -0,0 +1,31 @@
+# SQLAlchemy Models
+# Models for Schedule entities
+
+
+from sqlalchemy import (
+    Boolean,
+    Column,
+    DateTime,
+    ForeignKey,
+    Integer,
+    Sequence,
+    String,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import relationship
+
+from process_tracker.models.model_base import default_date, Base
+
+
+class ScheduleFrequency(Base):
+
+    __tablename__ = "schedule_frequency_lkup"
+    __table_args__ = {"schema": "process_tracker"}
+
+    schedule_frequency_id = Column(
+        Integer,
+        Sequence("schedule_frequency_schedule_frequency_id", schema="process_tracker"),
+        primary_key=True,
+        nullable=False,
+    )
+    schedule_frequency_name = Column(String(25), unique=True, nullable=False)

process_tracker/process_tracker.py

@@ -40,6 +40,7 @@
     ProcessTargetObjectAttribute,
     ProcessType,
 )
+from process_tracker.models.schedule import ScheduleFrequency
 from process_tracker.models.source import (
     DatasetType,
     Source,
@@ -67,6 +68,7 @@ def __init__(
         target_object_attributes=None,
         config_location=None,
         dataset_types=None,
+        schedule_frequency=None,
     ):
         """
         ProcessTracker is the primary engine for tracking data integration processes.
@@ -92,6 +94,8 @@ def __init__(
         :type config_location: file path
         :param dataset_types: A single dataset category type or list of dataset category types for the given process.
         :type dataset_types: list
+        :param schedule_frequency: The general scheduling frequency for the process (i.e. hourly)
+        :type schedule_frequency: string
         """
         self.config_location = config_location
         self.config = SettingsManager(config_location=self.config_location)
@@ -112,11 +116,21 @@ def __init__(
         )
         self.tool = self.data_store.get_or_create_item(model=Tool, tool_name=tool_name)
 
+        if schedule_frequency is None:
+            self.schedule_frequency = self.data_store.get_or_create_item(
+                model=ScheduleFrequency, schedule_frequency_name="unscheduled"
+            )
+        else:
+            self.schedule_frequency = self.data_store.get_or_create_item(
+                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,
         )
 
         # Dataset types should be loaded before source and target because they are also used there.
@@ -465,6 +479,25 @@ def find_process_contacts(self, process):
 
         return contacts
 
+    def find_process_by_schedule_frequency(self, frequency="daily"):
+        """
+        For the given schedule frequency, find the processed on that schedule frequency.
+        :param frequency: The desired frequency with which to retrieve processes.
+        :return:
+        """
+        process_list = list()
+
+        processes = (
+            self.data_store.session.query(Process.process_id)
+            .join(ScheduleFrequency)
+            .filter(ScheduleFrequency.schedule_frequency_name == frequency)
+        )
+
+        for process in processes:
+            process_list.append(process.process_id)
+
+        return process_list
+
     def get_latest_tracking_record(self, process):
         """
         For the given process, find the latest tracking record.

process_tracker/utilities/data_store.py

@@ -22,6 +22,7 @@
     ProcessType,
     ProcessStatus,
 )
+from process_tracker.models.schedule import ScheduleFrequency
 from process_tracker.models.source import Source
 from process_tracker.models.system import System
 from process_tracker.models.tool import Tool
@@ -38,7 +39,16 @@
 ]
 preload_process_status_types = ["running", "completed", "failed", "on hold"]
 preload_process_types = ["extract", "load"]
-preload_system_keys = [{"version", "0.6.0"}]
+preload_schedule_frequencies = [
+    "unscheduled",
+    "hourly",
+    "daily",
+    "weekly",
+    "monthly",
+    "quarterly",
+    "annually",
+]
+preload_system_keys = [{"version", "0.7.0"}]
 
 supported_data_stores = ["postgresql", "mysql", "oracle", "mssql", "snowflake"]
 
@@ -184,6 +194,13 @@ def initialize_data_store(self, overwrite=False):
             self.logger.info("Adding %s" % key)
             self.get_or_create_item(model=System, system_key=key, system_value=value)
 
+        self.logger.info("Adding schedule frequencies...")
+        for frequency in preload_schedule_frequencies:
+            self.logger.info("Adding %s" % frequency)
+            self.get_or_create_item(
+                model=ScheduleFrequency, schedule_frequency_name=frequency
+            )
+
         self.session.commit()
 
         self.logger.debug("Finished the initialization check.")

tests/test_process_tracker.py

@@ -1954,3 +1954,23 @@ def test_register_target_object_dataset_type(self):
 
         self.assertEqual(source_expected_result, source_given_result)
         self.assertEqual(object_expected_result, object_given_result)
+
+    def test_find_process_by_schedule_frequency(self):
+        """Testing that when querying based on a given frequency, the process id(s) associated with that frequency are returned."""
+
+        process = ProcessTracker(
+            process_name="Testing Register Target Object Dataset Type",
+            process_type="Extract",
+            actor_name="UnitTesting",
+            tool_name="Spark",
+            source_objects={"Unittests": ["Table1"]},
+            target_objects={"Unittest Target": ["Table1"]},
+            dataset_types="Category 1",
+            schedule_frequency="hourly",
+        )
+
+        given_result = process.find_process_by_schedule_frequency(frequency="hourly")
+
+        expected_result = [process.process.process_id]
+
+        self.assertEqual(expected_result, given_result)