Merge pull request #131 from OpenDataAlex/process_tracker_python-124

Process tracker python 124

Author: OpenDataAlex

dbscripts/mysql_process_tracker.sql

@@ -569,5 +569,49 @@ create table process_filter
 		foreign key (filter_type_id) references filter_type_lkup (filter_type_id)
 );
 
+create table process_tracker.extract_source
+(
+	extract_id int not null,
+	source_id int not null,
+	primary key (extract_id, source_id),
+	constraint extract_source_fk01
+		foreign key (extract_id) references process_tracker.extract_tracking (extract_id),
+	constraint extract_source_fk02
+		foreign key (source_id) references process_tracker.source_lkup (source_id)
+);
+
+create table process_tracker.extract_source_object
+(
+	extract_id int not null,
+	source_object_id int not null,
+	primary key (extract_id, source_object_id),
+	constraint extract_source_object_fk01
+		foreign key (extract_id) references process_tracker.extract_tracking (extract_id),
+	constraint extract_source_object_fk02
+		foreign key (source_object_id) references process_tracker.source_object_lkup (source_object_id)
+);
+
+create table process_tracker.source_location
+(
+	source_id int not null,
+	location_id int not null,
+	primary key (source_id, location_id),
+	constraint source_location_fk01
+		foreign key (source_id) references process_tracker.source_lkup (source_id),
+	constraint source_location_fk02
+		foreign key (location_id) references process_tracker.location_lkup (location_id)
+);
+
+create table process_tracker.source_object_location
+(
+	source_object_id int not null,
+	location_id int not null,
+	primary key (source_object_id, location_id),
+	constraint source_object_location_fk01
+		foreign key (source_object_id) references process_tracker.source_object_lkup (source_object_id),
+	constraint source_object_location_fk02
+		foreign key (location_id) references process_tracker.location_lkup (location_id)
+);
+
 
 

dbscripts/postgresql_process_tracker.sql

@@ -808,4 +808,61 @@ create table process_tracker.process_filter
 alter table process_tracker.process_filter owner to pt_admin;
 
 create unique index process_filter_udx01
-	on process_tracker.process_filter (process_id, source_object_attribute_id, filter_type_id);
+	on process_tracker.process_filter (process_id, source_object_attribute_id, filter_type_id);
+
+create table process_tracker.extract_source
+(
+	extract_id integer not null
+		constraint extract_source_fk02
+			references process_tracker.extract_tracking,
+	source_id integer not null
+		constraint extract_source_fk01
+			references process_tracker.source_lkup,
+	constraint extract_source_pk
+		primary key (extract_id, source_id)
+);
+
+alter table process_tracker.extract_source owner to pt_admin;
+
+create table process_tracker.extract_source_object
+(
+	extract_id integer not null
+		constraint extract_source_object_fk01
+			references process_tracker.extract_tracking,
+	source_object_id integer not null
+		constraint extract_source_object_fk02
+			references process_tracker.source_object_lkup,
+	constraint extract_source_object_pk
+		primary key (extract_id, source_object_id)
+);
+
+alter table process_tracker.extract_source_object owner to pt_admin;
+
+create table process_tracker.source_object_location
+(
+	source_object_id integer not null
+		constraint source_object_location_fk01
+			references process_tracker.source_object_lkup,
+	location_id integer not null
+		constraint source_object_location_fk02
+			references process_tracker.location_lkup,
+	constraint source_object_location_pk
+		primary key (source_object_id, location_id)
+);
+
+alter table process_tracker.source_object_location owner to pt_admin;
+
+create table process_tracker.source_location
+(
+	source_id integer not null
+		constraint source_location_fk01
+			references process_tracker.source_lkup,
+	location_id integer not null
+		constraint source_location_fk02
+			references process_tracker.location_lkup,
+	constraint source_location_pk
+		primary key (source_id, location_id)
+);
+
+alter table process_tracker.source_location owner to pt_admin;
+

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"