process_tracker_python-102 Add relationship to extract and extract location

✨ Added automated code to associate process source objects/sources
to any extracts used by the process.

Sources/Source Objects now are associated to extracts and the extract's
location.

Closes #102

Author: Alex Meadows

process_tracker/extract_tracker.py

@@ -17,8 +17,16 @@
     ExtractDatasetType,
     ExtractDependency,
     ExtractProcess,
+    ExtractSource,
+    ExtractSourceObject,
     ExtractStatus,
 )
+from process_tracker.models.source import (
+    Source,
+    SourceLocation,
+    SourceObjectAttribute,
+    SourceObjectLocation,
+)
 from process_tracker.models.source import DatasetType
 
 
@@ -104,6 +112,7 @@ def __init__(
             self.full_filename = self.get_full_filename()
             self.dataset_types = self.get_dataset_types()
             self.extract_process = self.retrieve_extract_process()
+            self.sources = self.extract.extract_sources
 
         else:
             if filename is None:
@@ -192,6 +201,28 @@ def __init__(
 
             self.extract_process = self.retrieve_extract_process()
 
+            if self.process_run.source_objects is not None:
+                self.logger.info(
+                    "Associating source system(s) object(s) with extract and location."
+                )
+                self.source_objects = self.register_extract_sources(
+                    source_objects=self.process_run.source_objects
+                )
+                self.sources = self.source_objects
+
+            elif self.process_run.process.sources is not None:
+                self.logger.info(
+                    "Associating source system(s) with extract and location."
+                )
+
+                self.sources = self.register_extract_sources(
+                    sources=self.process_run.sources
+                )
+
+            else:
+                self.logger.info("No source system(s) to associate to.")
+                self.sources = None
+
             if status is not None:
                 self.logger.info("Status was provided by user.")
                 self.change_extract_status(new_status=status)
@@ -422,6 +453,60 @@ def register_extract_dataset_types(self, dataset_types):
 
         return dataset_types
 
+    def register_extract_sources(self, sources=None, source_objects=None):
+        """
+        For the provided sources from process_run instance, associate with given Extract instance.
+        :param sources: List of sources from process_run record.
+        :param source_objects: List of sources and their objects from process_run record.
+        :return:
+        """
+        source_list = list()
+
+        if source_objects is not None:
+
+            for object in source_objects:
+                self.logger.debug(
+                    "Associating extract %s to source %s."
+                    % (self.extract.extract_id, object.source_object_id)
+                )
+
+                source_object = self.data_store.get_or_create_item(
+                    model=ExtractSourceObject,
+                    extract_id=self.extract.extract_id,
+                    source_object_id=object.source_object_id,
+                )
+                source_list.append(source_object)
+
+                self.data_store.get_or_create_item(
+                    SourceObjectLocation,
+                    source_object_id=object.source_object_id,
+                    location_id=self.extract.extract_location_id,
+                )
+
+        elif sources is not None:
+
+            for source in sources:
+
+                self.logger.debug(
+                    "Associating extract %s to source %s."
+                    % (self.extract.extract_id, source.source_id)
+                )
+                extract_source = self.data_store.get_or_create_item(
+                    model=ExtractSource,
+                    extract_id=self.extract.extract_id,
+                    source_id=source.source_id,
+                )
+                source_list.append(extract_source)
+                self.logger.debug("Extract source record created. %s" % extract_source)
+
+                self.data_store.get_or_create_item(
+                    model=SourceLocation,
+                    source_id=source.source_id,
+                    location_id=self.extract.extract_location_id,
+                )
+
+        return source_list
+
     def retrieve_extract_process(self):
         """
         Create and initialize or retrieve the process/extract relationship.

process_tracker/models/extract.py

@@ -139,6 +139,7 @@ class Extract(Base):
     extract_process = relationship(
         "ExtractProcess", back_populates="process_extracts", passive_deletes="all"
     )
+    extract_sources = relationship("ExtractSource")
     extract_status = relationship("ExtractStatus", foreign_keys=[extract_status_id])
     locations = relationship("Location", foreign_keys=[extract_location_id])
 
@@ -254,6 +255,68 @@ def __repr__(self):
         )
 
 
+class ExtractSource(Base):
+
+    __tablename__ = "extract_source"
+    __table_args__ = {"schema": "process_tracker"}
+
+    extract_id = Column(
+        Integer,
+        ForeignKey("process_tracker.extract_tracking.extract_id"),
+        primary_key=True,
+        nullable=False,
+    )
+    source_id = Column(
+        Integer,
+        ForeignKey("process_tracker.source_lkup.source_id"),
+        primary_key=True,
+        nullable=False,
+    )
+
+    UniqueConstraint(extract_id, source_id)
+
+    extracts = relationship("Extract")
+    sources = relationship("Source")
+
+    def __repr__(self):
+
+        return "<ExtractSource extract_id=%s, source_id=%s>" % (
+            self.extract_id,
+            self.source_id,
+        )
+
+
+class ExtractSourceObject(Base):
+
+    __tablename__ = "extract_source_object"
+    __table_args__ = {"schema": "process_tracker"}
+
+    extract_id = Column(
+        Integer,
+        ForeignKey("process_tracker.extract_tracking.extract_id"),
+        primary_key=True,
+        nullable=False,
+    )
+    source_object_id = Column(
+        Integer,
+        ForeignKey("process_tracker.source_object_lkup.source_object_id"),
+        primary_key=True,
+        nullable=False,
+    )
+
+    UniqueConstraint(extract_id, source_object_id)
+
+    extracts = relationship("Extract")
+    source_objects = relationship("SourceObject")
+
+    def __repr__(self):
+
+        return "<ExtractSourceObject extract_id=%s, source_object_id=%s>" % (
+            self.extract_id,
+            self.source_object_id,
+        )
+
+
 class LocationType(Base):
 
     __tablename__ = "location_type_lkup"

process_tracker/models/source.py

@@ -153,6 +153,37 @@ def __repr__(self):
         )
 
 
+class SourceLocation(Base):
+
+    __tablename__ = "source_location"
+    __table_args__ = {"schema": "process_tracker"}
+
+    source_id = Column(
+        Integer,
+        ForeignKey("process_tracker.source_lkup.source_id"),
+        nullable=False,
+        primary_key=True,
+    )
+    location_id = Column(
+        Integer,
+        ForeignKey("process_tracker.location_lkup.location_id"),
+        nullable=False,
+        primary_key=True,
+    )
+
+    UniqueConstraint(source_id, location_id)
+
+    sources = relationship("Source")
+    locations = relationship("Location")
+
+    def __repr__(self):
+
+        return "<SourceLocation source_id=%s, location_id=%s>" % (
+            self.source_id,
+            self.location_id,
+        )
+
+
 class SourceObject(Base):
 
     __tablename__ = "source_object_lkup"
@@ -194,6 +225,35 @@ def __repr__(self):
         )
 
 
+class SourceObjectLocation(Base):
+    __tablename__ = "source_object_location"
+    __table_args__ = {"schema": "process_tracker"}
+
+    source_object_id = Column(
+        Integer,
+        ForeignKey("process_tracker.source_object_lkup.source_object_id"),
+        nullable=False,
+        primary_key=True,
+    )
+    location_id = Column(
+        Integer,
+        ForeignKey("process_tracker.location_lkup.location_id"),
+        nullable=False,
+        primary_key=True,
+    )
+
+    UniqueConstraint(source_object_id, location_id)
+
+    source_objects = relationship("SourceObject")
+    locations = relationship("Location")
+
+    def __repr__(self):
+        return "<SourceObjectLocation source_object_id=%s, location_id=%s>" % (
+            self.source_object_id,
+            self.location_id,
+        )
+
+
 class SourceObjectAttribute(Base):
 
     __tablename__ = "source_object_attribute_lkup"
@@ -202,7 +262,7 @@ class SourceObjectAttribute(Base):
     source_object_attribute_id = Column(
         Integer,
         Sequence(
-            "source_object_attribute_lkup_source_object_attribute_id_seq",
+            "source_object_attribute_source_object_attribute_id_seq",
             schema="process_tracker",
         ),
         primary_key=True,

process_tracker/process_tracker.py

@@ -196,14 +196,19 @@ def __init__(
 
             # sources, source_objects, or source_object_attributes should be set, not multiple.  Always go with lower grain if possible.
 
+            self.source_object_attributes = None
+            self.source_objects = None
+
             if source_object_attributes is not None:
-                self.sources = self.register_process_sources(
+                self.source_object_attributes = self.register_process_sources(
                     source_object_attributes=source_object_attributes
                 )
+                self.sources = self.source_object_attributes
             elif source_objects is not None:
-                self.sources = self.register_process_sources(
+                self.source_objects = self.register_process_sources(
                     source_objects=source_objects
                 )
+                self.sources = self.source_objects
             elif sources is not None:
                 self.sources = self.register_process_sources(sources=sources)
             else: