Added ExtractDependency Table

Added new table for handling extract dependencies and method to check for
dependencies which is used when an extract is attempting to switch to
'loading' status.

Started working on #14

Author: OpenDataAlex

Pipfile

@@ -21,6 +21,7 @@ snowflake-sqlalchemy = "*"
 pymysql = "*"
 cx-oracle = "*"
 pymssql = "*"
+black = "*"
 
 [requires]
 python_version = "3.7"

process_tracker/extract_tracker.py

@@ -2,14 +2,14 @@
 # Used in the creation and editing of extract records.  Used in conjunction with process tracking.
 from datetime import datetime
 import logging
-import os
 from os.path import join
 
+from sqlalchemy.orm import aliased
 
 from process_tracker.data_store import DataStore
 from process_tracker.location_tracker import LocationTracker
 from process_tracker.utilities.settings import SettingsManager
-from process_tracker.models.extract import Extract, ExtractProcess, ExtractStatus, Location
+from process_tracker.models.extract import Extract, ExtractDependency, ExtractProcess, ExtractStatus
 
 
 class ExtractTracker:
@@ -97,6 +97,11 @@ def change_extract_status(self, new_status):
         """
         status_date = datetime.now()
         if new_status in self.extract_status_types:
+
+            if new_status == self.extract_status_loading:
+
+                self.extract_dependency_check()
+
             self.logger.info('Setting extract status to %s' % new_status)
 
             new_status = self.extract_status_types[new_status]
@@ -113,6 +118,29 @@ def change_extract_status(self, new_status):
             raise Exception('%s is not a valid extract status type.  '
                             'Please add the status to extract_status_lkup' % new_status)
 
+    def extract_dependency_check(self):
+        """
+        Determine if the extract file has any unloaded dependencies before trying to load the file.
+        :return:
+        """
+        child_extract = aliased(Extract)
+        parent_extract = aliased(Extract)
+
+        dependency_hold = self.session.query(ExtractDependency) \
+                                .join(parent_extract, ExtractDependency.parent_extract_id == parent_extract.extract_id) \
+                                .join(child_extract, ExtractDependency.child_extract_id == child_extract.extract_id) \
+                                .join(Extract, Extract.extract_id == parent_extract.extract_id) \
+                                .join(ExtractStatus, ExtractStatus.extract_status_id == Extract.extract_status_id) \
+                                .filter(child_extract.extract_id == self.extract.extract_id) \
+                                .filter(ExtractStatus.extract_status_name.in_(('loading', 'initializing', 'ready'))) \
+                                .count()
+
+        if dependency_hold > 0:
+            self.logger.error('Extract files that this extract file is dependent on have not been loaded, are being '
+                              'created, or are in the process of loading.')
+            raise Exception('Extract files that this extract file is dependent on have not been loaded, are being '
+                            'created, or are in the process of loading.')
+
     def get_extract_status_types(self):
         """
         Get list of process status types and return dictionary.

process_tracker/models/extract.py

@@ -50,6 +50,22 @@ def full_filepath(self):
         return join(self.locations.location_path, self.extract_filename)
 
 
+class ExtractDependency(Base):
+
+    __tablename__ = 'extract_dependency'
+
+    parent_extract_id = Column(Integer, ForeignKey('extract_tracking.extract_id'), primary_key=True)
+    child_extract_id = Column(Integer, ForeignKey('extract_tracking.extract_id'), primary_key=True)
+
+    child_extract = relationship('Extract', foreign_keys=[child_extract_id])
+    parent_extract = relationship('Extract', foreign_keys=[parent_extract_id])
+
+    def __repr__(self):
+
+        return "<ExtractDependency (parent_extract=%s, child_extract=%s)>" % (self.parent_extract_id
+                                                                              , self.child_extract_id)
+
+
 class ExtractProcess(Base):
 
     __tablename__ = "extract_process_tracking"

process_tracker/models/system.py

@@ -16,4 +16,4 @@ class System(Base):
 
     def __repr__(self):
 
-        return "<System (system_key=%s)>" % self.system_key
+        return "<System (system_key=%s)>" % self.system_key

setup.py

@@ -31,6 +31,7 @@
     ],
     extras_requires={
         'dev': [
+            'black >= 19.3b0',
             'coverage >= "4.0.3',
             'coveralls >= 1.7.0',
             'moto >= 1.3.8',

tests/test_extract_tracker.py

@@ -153,6 +153,20 @@ def test_derive_location_name_s3(self):
 
         self.assertEqual(expected_result, given_result)
 
+    def test_extract_dependency_check(self):
+        """
+        Testing that if no dependencies are in a state that doesn't stop an extract from being loaded, then the extract
+        is loaded.
+        :return:
+        """
+
+    def test_extract_dependency_check_blocked(self):
+        """
+        Testing that if a dependency is in a state that stops an extract from being loaded, then the extract triggers an
+        error blocking the file from being processed.
+        :return:
+        """
+
     def test_location_name_provided(self):
         """
         Testing that if a location name is provided (like with default extract), one is not created.