process_tracker_python-50 Extract Finders By Given Status

🐛 Hit connection threshold to database
✨ Extract finders can now search for any status type.

Tweaked ExtractTracker to use the data store and session objects from process_tracker, since an extract tracker must have a process tracker
associated with it.  This stopped the connection threshold problem as
the same connection can be used until the process tracker changes to 'completed' or 'failed' status, which is when the session is closed.

Finder methods now have a default status of 'ready' (which should be the
common status required) but any status type can be searched for, which
allows custom status types to be leveraged for extracts.

Closes #50

Author: OpenDataAlex

.gitignore

@@ -173,4 +173,3 @@ pip-selfcheck.json
 .idea/misc.xml
 .idea/modules.xml
 .idea/process_tracker_python.iml
-/tests/test_process_tracker.py

process_tracker/extract_tracker.py

@@ -49,10 +49,11 @@ def __init__(
         self.logger = logging.getLogger(__name__)
         self.logger.setLevel(config["DEFAULT"]["log_level"])
 
-        self.data_store = DataStore()
-        self.session = self.data_store.session
         self.process_run = process_run
 
+        self.data_store = self.process_run.data_store
+        self.session = self.process_run.session
+
         self.filename = filename
 
         if location is not None:

process_tracker/models/extract.py

@@ -60,6 +60,7 @@ class Extract(Base):
     extract_load_record_count = Column(Integer, nullable=True)
 
     extract_process = relationship("ExtractProcess", back_populates="process_extracts")
+    extract_status = relationship("ExtractStatus", foreign_keys=[extract_status_id])
     locations = relationship("Location", foreign_keys=[extract_location_id])
 
     def __repr__(self):

process_tracker/process_tracker.py

@@ -150,22 +150,29 @@ def change_run_status(self, new_status, end_date=None):
 
             self.session.commit()
 
+            if (
+                self.process_status_types[new_status] == self.process_status_complete
+            ) or (self.process_status_types[new_status] == self.process_status_failed):
+                self.session.close()
+
         else:
             raise Exception("The provided status type %s is invalid." % new_status)
 
-    def find_ready_extracts_by_filename(self, filename):
+    def find_extracts_by_filename(self, filename, status="ready"):
         """
         For the given filename, or filename part, find all matching extracts that are ready for processing.
-
-        :param filename:
-        :return:
+        :param filename: Filename or part of filename.
+        :type filename: str
+        :param status: Name of the status type for files being searched.  Default 'ready'.
+        :type status: str
+        :return: List of Extract SQLAlchemy objects.
         """
 
         process_files = (
             self.session.query(Extract)
             .join(ExtractStatus)
             .filter(Extract.extract_filename.like("%" + filename + "%"))
-            .filter(ExtractStatus.extract_status_name == "ready")
+            .filter(ExtractStatus.extract_status_name == status)
             .order_by(Extract.extract_registration_date_time)
             .order_by(Extract.extract_id)
             .all()
@@ -175,22 +182,26 @@ def find_ready_extracts_by_filename(self, filename):
 
         return process_files
 
-    def find_ready_extracts_by_location(self, location_name=None, location_path=None):
+    def find_extracts_by_location(
+        self, location_name=None, location_path=None, status="ready"
+    ):
         """
         For the given location path or location name, find all matching extracts that are ready for processing
         :param location_name: The name of the location
         :type location_name: str
         :param location_path: The path of the location
         :type location_path: str
-        :return: List of extract files that are in 'ready' state'.
+        :param status: Name of the status type for files being searched.  Default 'ready'.
+        :type status: str
+        :return: List of Extract SQLAlchemy objects.
         """
 
         if location_path is not None:
             process_files = (
                 self.session.query(Extract)
                 .join(Location)
                 .join(ExtractStatus)
-                .filter(ExtractStatus.extract_status_name == "ready")
+                .filter(ExtractStatus.extract_status_name == status)
                 .filter(Location.location_path == location_path)
                 .order_by(Extract.extract_registration_date_time)
                 .order_by(Extract.extract_id)
@@ -201,7 +212,7 @@ def find_ready_extracts_by_location(self, location_name=None, location_path=None
                 self.session.query(Extract)
                 .join(Location)
                 .join(ExtractStatus)
-                .filter(ExtractStatus.extract_status_name == "ready")
+                .filter(ExtractStatus.extract_status_name == status)
                 .filter(Location.location_name == location_name)
                 .order_by(Extract.extract_registration_date_time)
                 .order_by(Extract.extract_id)
@@ -218,10 +229,14 @@ def find_ready_extracts_by_location(self, location_name=None, location_path=None
         self.logger.info("Returning extract files by location.")
         return process_files
 
-    def find_ready_extracts_by_process(self, extract_process_name):
+    def find_extracts_by_process(self, extract_process_name, status="ready"):
         """
         For the given named process, find the extracts that are ready for processing.
-        :return: List of OS specific filepaths with filenames.
+        :param extract_process_name: Name of the process that is associated with extracts
+        :type extract_process_name: str
+        :param status: Name of the status type for files being searched.  Default 'ready'.
+        :type status: str
+        :return: List of Extract SQLAlchemy objects.
         """
 
         process_files = (
@@ -238,7 +253,7 @@ def find_ready_extracts_by_process(self, extract_process_name):
             .join(Process)
             .filter(
                 Process.process_name == extract_process_name,
-                ExtractStatus.extract_status_name == "ready",
+                ExtractStatus.extract_status_name == status,
             )
             .order_by(Extract.extract_registration_date_time)
             .all()

tests/test_extract_tracker.py

@@ -46,6 +46,7 @@ def tearDownClass(cls):
         cls.session.query(ProcessDependency).delete()
         cls.session.query(Process).delete()
         cls.session.commit()
+        cls.session.close()
 
     def setUp(self):
         """