process_tracker_python-7 Register extracts by location

✨ Can now register extracts by given location
🐛 Have been unable to write test for s3 locations.

Tried writing test for s3 locations to register, but having issues with
moto library.  Waiting to hear back from stackoverflow, but wanted to at
least commit what I've done before switching to another issue in the
meantime.

Author: OpenDataAlex

process_tracker/location_tracker.py

@@ -14,13 +14,13 @@ def __init__(self, location_path, location_name=None):
         self.data_store = DataStore()
 
         self.location_path = location_path.lower()
-        self.location_type = self.derive_location_type()
 
         if location_name is None:
             self.location_name = self.derive_location_name()
         else:
             self.location_name = location_name
 
+        self.location_type = self.derive_location_type()
         self.location = self.data_store.get_or_create(model=Location
                                                       , location_name=self.location_name
                                                       , location_path=location_path
@@ -56,7 +56,7 @@ def derive_location_type(self):
         :return:
         """
 
-        if "s3" in self.location_path:
+        if "s3" in self.location_path or "s3" in self.location_name:
 
             location_type = self.data_store.get_or_create(model=LocationType
                                                           , location_type_name="s3")

process_tracker/process_tracker.py

@@ -228,9 +228,15 @@ def register_extracts_by_location(self, location_path, location_name=None):
         """
         location = LocationTracker(location_path=location_path, location_name=location_name)
 
-        if location.location_type == "s3":
-            s3 = boto3.resource('s3')
-            bucket = s3.Bucket(location.location_path)
+        if location.location_type.location_type_name == "s3":
+            s3 = boto3.resource("s3")
+
+            path = location.location_path
+
+            if path.startswith("s3://"):
+                path = path[len("s3://")]
+
+            bucket = s3.Bucket(path)
 
             for file in bucket.objects.all():
                 ExtractTracker(process_run=self

tests/test_process_tracker.py

@@ -3,6 +3,7 @@
 from datetime import datetime, timedelta
 import unittest
 
+import boto3
 import moto
 from sqlalchemy.orm import aliased, Session
 
@@ -316,13 +317,40 @@ def test_register_extracts_by_location_local(self):
 
         self.assertEqual(expected_result, given_result)
 
-    def test_register_extracts_by_location_s3(self):
-        """
-        Testing that when the location is s3, all the extracts are registered and set to 'ready' status.
-        :return:
-        """
-
-
+    # def test_register_extracts_by_location_s3(self):
+    #     """
+    #     Testing that when the location is s3, all the extracts are registered and set to 'ready' status.
+    #     The process/extract relationship should also be set to 'ready' since that is the last status the process set
+    #     the extracts to.
+    #     :return:
+    #     """
+    #     process_status = aliased(ExtractStatus)
+    #     extract_status = aliased(ExtractStatus)
+    #
+    #     expected_keys = 'test_local_dir_1.csv', 'test_local_dir_2.csv'
+    #
+    #     with moto.mock_s3():
+    #         conn = boto3.resource('s3', region_name='us-east-1')
+    #         conn.create_bucket(Bucket='test_bucket')
+    #
+    #         for file in expected_keys:
+    #             conn.Object('test_bucket', file)
+    #
+    #         self.process_tracker.register_extracts_by_location(location_path='s3://test_bucket')
+    #
+    #     extracts = self.session.query(Extract.extract_filename, extract_status.extract_status_name, process_status.extract_status_name)\
+    #                            .join(ExtractProcess, Extract.extract_id == ExtractProcess.extract_tracking_id) \
+    #                            .join(extract_status, Extract.extract_status_id == extract_status.extract_status_id) \
+    #                            .join(process_status, ExtractProcess.extract_process_status_id == process_status.extract_status_id) \
+    #                            .filter(ExtractProcess.process_tracking_id == self.process_tracker.process_tracking_run.process_tracking_id)
+    #
+    #     given_result = [[extracts[0].extract_filename, extracts[0].extract_status_name, extracts[0].extract_status_name]
+    #                     ,[extracts[1].extract_filename, extracts[1].extract_status_name, extracts[1].extract_status_name]]
+    #
+    #     expected_result = [['test_local_dir_1.csv', 'ready', 'ready']
+    #                       ,['test_local_dir_2.csv', 'ready', 'ready']]
+    #
+    #     self.assertEqual(expected_result, given_result)
 
     def test_register_new_process_run(self):
         """