process_tracker_python-9 Process can have more than one source

OpenDataAlex · OpenDataAlex · commit f378eef98814 · 2019-05-18T15:58:25.000-04:00
✨ Process now can have more than one source.  Can also have more than one target

Processes now have the ability to be registered with more than one source
and more than one target (which is an alias of source).
diff --git a/dbscripts/postgresql_process_tracker.sql b/dbscripts/postgresql_process_tracker.sql
@@ -4,28 +4,13 @@ create schema process_tracking;
 
 alter schema process_tracking owner to pt_admin;
 
-create sequence actor_lkup_actor_id_seq;
-
-alter sequence actor_lkup_actor_id_seq owner to pt_admin;
-
-create table location_lkup
-(
-	location_id serial not null
-		constraint location_lkup_pk
-			primary key,
-	location_name varchar(750) not null,
-	location_path varchar(750) not null
-);
-
-comment on table location_lkup is 'Locations where files are located.';
+create schema process_tracking;
 
-alter table location_lkup owner to pt_admin;
+alter schema process_tracking owner to pt_admin;
 
-create unique index location_lkup_udx01
-	on location_lkup (location_name);
+create sequence actor_lkup_actor_id_seq;
 
-create unique index location_lkup_udx02
-	on location_lkup (location_path);
+alter sequence actor_lkup_actor_id_seq owner to pt_admin;
 
 create table error_type_lkup
 (
@@ -136,9 +121,6 @@ create table process
 		constraint process_pk
 			primary key,
 	process_name varchar(250) not null,
-	process_source_id integer not null
-		constraint process_fk01
-			references source_lkup,
 	total_record_count integer default 0 not null,
 	process_type_id integer not null
 		constraint process_fk02
@@ -153,8 +135,6 @@ comment on table process is 'Processes being tracked';
 
 comment on column process.process_name is 'Unique name for process.';
 
-comment on column process.process_source_id is 'The source that the process is extracting from.';
-
 comment on column 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.';
@@ -274,12 +254,45 @@ alter table extract_status_lkup owner to pt_admin;
 create unique index extract_status_lkup_extract_status_name_uindex
 	on extract_status_lkup (extract_status_name);
 
+create table location_type_lkup
+(
+	location_type_id serial not null
+		constraint location_type_lkup_pk
+			primary key,
+	location_type_name varchar(25) not null
+);
+
+comment on table location_type_lkup is 'Listing of location types';
+
+alter table location_type_lkup owner to pt_admin;
+
+create table location_lkup
+(
+	location_id serial not null
+		constraint location_lkup_pk
+			primary key,
+	location_name varchar(750) not null,
+	location_path varchar(750) not null,
+	location_type integer not null
+		constraint location_lkup_fk01
+			references location_type_lkup
+);
+
+comment on table location_lkup is 'Locations where files are located.';
+
+alter table location_lkup owner to pt_admin;
+
+create unique index location_lkup_udx01
+	on location_lkup (location_name);
+
+create unique index location_lkup_udx02
+	on location_lkup (location_path);
+
 create table extract_tracking
 (
 	extract_id serial not null
 		constraint extract_tracking_pk
 			primary key,
-	extract_source_id integer not null,
 	extract_filename varchar(750) not null,
 	extract_location_id integer not null
 		constraint extract_tracking_fk01
@@ -295,8 +308,6 @@ create table extract_tracking
 
 comment on table extract_tracking is 'Tracking table for all extract/staging data files.';
 
-comment on column extract_tracking.extract_source_id is 'Source identifier (source_lkup) for where the extract originated.';
-
 comment on column extract_tracking.extract_filename is 'The unique filename for a given extract from a given source.';
 
 comment on column extract_tracking.extract_location_id is 'The location where the given extract can be found.';
@@ -309,9 +320,6 @@ comment on column extract_tracking.extract_registration_date_time is 'The dateti
 
 alter table extract_tracking owner to pt_admin;
 
-create unique index extract_tracking_udx01
-	on extract_tracking (extract_source_id, extract_filename);
-
 create table extract_process_tracking
 (
 	extract_tracking_id integer not null
@@ -332,6 +340,43 @@ comment on table extract_process_tracking is 'Showing which processes have impac
 
 alter table extract_process_tracking owner to pt_admin;
 
+create unique index location_type_lkup_udx01
+	on location_type_lkup (location_type_name);
+
+create table process_source
+(
+	source_id integer not null
+		constraint process_source_fk01
+			references source_lkup,
+	process_id integer not null
+		constraint process_source_fk02
+			references process,
+	constraint process_source_pk
+		primary key (source_id, process_id)
+);
+
+comment on table process_source is 'List of sources for given processes';
+
+alter table process_source owner to pt_admin;
+
+create table process_target
+(
+	target_source_id integer not null
+		constraint process_target_fk01
+			references source_lkup,
+	process_id integer not null
+		constraint process_target_fk02
+			references process,
+	constraint process_target_pk
+		primary key (target_source_id, process_id)
+);
+
+comment on table process_target is 'List of targets for given processes';
+
+alter table process_target owner to pt_admin;
+
+
+
 INSERT INTO process_tracking.extract_status_lkup (extract_status_id, extract_status_name) VALUES (1, 'initializing');
 INSERT INTO process_tracking.extract_status_lkup (extract_status_id, extract_status_name) VALUES (2, 'ready');
 INSERT INTO process_tracking.extract_status_lkup (extract_status_id, extract_status_name) VALUES (3, 'loading');
diff --git a/process_tracker/extract_tracker.py b/process_tracker/extract_tracker.py
@@ -36,7 +36,6 @@ def __init__(self, process_run, filename, location=None, location_name=None, loc
         self.session = self.data_store.session
         self.process_run = process_run
 
-        self.source = self.process_run.source
         self.filename = filename
 
         if location is not None:
@@ -48,8 +47,7 @@ def __init__(self, process_run, filename, location=None, location_name=None, loc
 
         self.extract = self.data_store.get_or_create(model=Extract
                                                      , extract_filename=filename
-                                                     , extract_location_id=self.location.location.location_id
-                                                     , extract_source_id=self.source.source_id)
+                                                     , extract_location_id=self.location.location.location_id)
 
         if location_path is not None:
             self.full_filename = join(location_path, filename)
diff --git a/process_tracker/models/extract.py b/process_tracker/models/extract.py
@@ -24,14 +24,10 @@ class Extract(Base):
     __tablename__ = "extract_tracking"
 
     extract_id = Column(Integer, Sequence('extract_tracking_extract_id_seq'), primary_key=True)
-    extract_source_id = Column(Integer, ForeignKey("source_lkup.source_id"))
     extract_filename = Column(String(750), nullable=False, unique=True)
     extract_location_id = Column(Integer, ForeignKey('location_lkup.location_id'))
-#    extract_process_run_id = Column(Integer, ForeignKey('process_tracking.process_tracking_id'))
     extract_status_id = Column(Integer, ForeignKey('extract_status_lkup.extract_status_id'))
     extract_registration_date_time = Column(DateTime, nullable=False, default=datetime.now())
-#    extract_load_date_time = Column(DateTime, nullable=False, default=default_date)
-#    extract_archive_date_time = Column(DateTime, nullable=False, default=default_date)
 
     extract_process = relationship("ExtractProcess", back_populates='process_extracts')
     locations = relationship("Location", foreign_keys=[extract_location_id])
diff --git a/process_tracker/models/process.py b/process_tracker/models/process.py
@@ -75,31 +75,52 @@ class Process(Base):
 
     process_id = Column(Integer, Sequence('process_process_id_seq'), primary_key=True)
     process_name = Column(String(250), nullable=False, unique=True)
-    process_source_id = Column(Integer, ForeignKey('source_lkup.source_id'))
-#    latest_run_low_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-#    latest_run_high_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-#    latest_run_id = Column(Integer, nullable=False, default=0)
-#    latest_run_start_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-#    latest_run_end_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
-#    latest_run_process_status = Column(Integer, nullable=False, default=0)
-#    latest_run_record_count = Column(Integer, nullable=False, default=0)
     total_record_count = Column(Integer, nullable=False, default=0)
-#    latest_run_actor_id = Column(id.idType, ForeignKey('actor.actor_id'))
     process_type_id = Column(Integer, ForeignKey('process_type_lkup.process_type_id'))
     process_tool_id = Column(Integer, ForeignKey('tool_lkup.tool_id'))
     last_failed_run_date_time = Column(DateTime(timezone=True), nullable=False, default=default_date)
 
     process_tracking = relationship("ProcessTracking")
     process_type = relationship("ProcessType", back_populates="processes")
-    source = relationship("Source")
+    sources = relationship("ProcessSource")
+    targets = relationship("ProcessTarget")
     tool = relationship("Tool")
 
     def __repr__(self):
 
-        return "<Process (id=%s, name=%s, source=%s, type=%s)>" % (self.process_id
-                                                                     , self.process_name
-                                                                     , self.process_source_id
-                                                                     , self.process_type_id)
+        return "<Process (id=%s, name=%s, type=%s)>" % (self.process_id
+                                                        , self.process_name
+                                                        , self.process_type_id)
+
+
+class ProcessSource(Base):
+
+    __tablename__ = 'process_source'
+
+    source_id = Column(Integer, ForeignKey('source_lkup.source_id'), primary_key=True)
+    process_id = Column(Integer, ForeignKey('process.process_id'), primary_key=True)
+
+    sources = relationship("Source")
+    processes = relationship("Process")
+
+    def __repr__(self):
+
+        return "<ProcessSource (process=%s, source=%s)>" % (self.process_id
+                                                                   , self.source_id)
+
+
+class ProcessTarget(Base):
+    __tablename__ = 'process_target'
+
+    target_source_id = Column(Integer, ForeignKey('source_lkup.source_id'), primary_key=True)
+    process_id = Column(Integer, ForeignKey('process.process_id'), primary_key=True)
+
+    targets = relationship("Source")
+    processes = relationship("Process")
+
+    def __repr__(self):
+        return "<ProcessSource (process=%s, target_source=%s)>" % (self.process_id
+                                                                   , self.source_id)
 
 
 class ProcessDependency(Base):
diff --git a/process_tracker/process_tracker.py b/process_tracker/process_tracker.py
@@ -14,20 +14,21 @@
 
 from process_tracker.models.actor import Actor
 from process_tracker.models.extract import Extract, ExtractProcess, ExtractStatus, Location
-from process_tracker.models.process import ErrorTracking, ErrorType, Process, ProcessTracking, ProcessStatus, ProcessType
+from process_tracker.models.process import ErrorTracking, ErrorType, Process, ProcessTracking, ProcessStatus, ProcessSource, ProcessTarget, ProcessType
 from process_tracker.models.source import Source
 from process_tracker.models.tool import Tool
 
 
 class ProcessTracker:
 
-    def __init__(self, process_name, process_type, actor_name, tool_name, source_name):
+    def __init__(self, process_name, process_type, actor_name, tool_name, sources, targets):
         """
         ProcessTracker is the primary engine for tracking data integration processes.
         :param process_name: Name of the process being tracked.
         :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 source_name: Name of the source that the data is coming from.
+        :param sources: A single source name or list of source names for the given process.
+        :type sources: list
         """
 
         self.logger = logging.getLogger(__name__)
@@ -37,14 +38,15 @@ def __init__(self, process_name, process_type, actor_name, tool_name, source_nam
 
         self.actor = self.data_store.get_or_create(model=Actor, actor_name=actor_name)
         self.process_type = self.data_store.get_or_create(model=ProcessType, process_type_name=process_type)
-        self.source = self.data_store.get_or_create(model=Source, source_name=source_name)
         self.tool = self.data_store.get_or_create(model=Tool, tool_name=tool_name)
 
         self.process = self.data_store.get_or_create(model=Process, process_name=process_name
-                                                     , process_source_id=self.source.source_id
                                                      , process_type_id=self.process_type.process_type_id
                                                      , process_tool_id=self.tool.tool_id)
 
+        self.sources = self.register_process_sources(sources=sources)
+        self.targets = self.register_process_targets(targets=targets)
+
         self.process_name = process_name
 
         # Getting all status types in the event there are custom status types added later.
@@ -289,6 +291,46 @@ def register_new_process_run(self):
         else:
             raise Exception('The process %s is currently running.' % self.process_name)
 
+    def register_process_sources(self, sources):
+        """
+        Register source(s) to a given process.
+        :param sources: List of source name(s)
+        :return: List of source objects.
+        """
+        if sources != list:
+            sources = [sources]
+
+        source_list = []
+
+        for source in sources:
+            source = self.data_store.get_or_create(model=Source, source_name=source)
+
+            self.data_store.get_or_create(model=ProcessSource, source_id=source.source_id
+                                          , process_id=self.process.process_id)
+
+            source_list.append(source)
+        return source_list
+
+    def register_process_targets(self, targets):
+        """
+        Register target source(s) to a given process.
+        :param targets: List of source name(s)
+        :return: List of source objects.
+        """
+        if targets != list:
+            targets = [targets]
+
+        source_list = []
+
+        for target in targets:
+            source = self.data_store.get_or_create(model=Source, source_name=target)
+
+            self.data_store.get_or_create(model=ProcessTarget, target_source_id=source.source_id
+                                          , process_id=self.process.process_id)
+
+            source_list.append(source)
+        return source_list
+
     def set_process_run_low_high_dates(self, low_date=None, high_date=None):
         """
         For the given process run, set the process_run_low_date_time and/or process_run_high_date_time.
diff --git a/tests/test_extract_tracker.py b/tests/test_extract_tracker.py
@@ -3,7 +3,7 @@
 import unittest
 
 from process_tracker.models.extract import Extract, ExtractProcess, Location
-from process_tracker.models.process import Process, ProcessTracking
+from process_tracker.models.process import Process, ProcessSource, ProcessTarget, ProcessTracking
 
 from process_tracker.data_store import DataStore
 from process_tracker.extract_tracker import ExtractTracker
@@ -18,7 +18,8 @@ def setUpClass(cls):
                                              , process_type='Load'
                                              , actor_name='UnitTesting'
                                              , tool_name='Spark'
-                                             , source_name='Unittests')
+                                             , sources='Unittests'
+                                             , targets='Unittests')
 
         cls.process_run = cls.process_tracker
 
@@ -29,6 +30,8 @@ def setUpClass(cls):
     def tearDownClass(cls):
 
         cls.session.query(ProcessTracking).delete()
+        cls.session.query(ProcessSource).delete()
+        cls.session.query(ProcessTarget).delete()
         cls.session.query(Process).delete()
         cls.session.commit()
 
diff --git a/tests/test_process_tracker.py b/tests/test_process_tracker.py
