Fix list_objects and get_most_recent_object_key methods

The `list_objects` method now returns a list of each object's metadata. The `get_most_recent_object_key` method now returns the key of the object with the most recent `LastModified` datetime.

Author: oliverjwroberts

datahub/ingest/boto3.py

@@ -43,17 +43,31 @@ def __init__(
             self.s3_client = s3_client
 
     def list_objects(self) -> list[str]:
-        """Returns a list of all objects with specified prefix."""
+        """Returns a list of metadata about each object located at self.prefix.
+
+        From the boto3 docs, the metadata will contain the following:
+        - Key (string) - the name that you assign the object
+        - LastModified (datetime) - creation date of the object
+        - ETag (string) - entity tag; a hash of the object
+        - ChecksumAlgorithm (list) - the algorithm used to create a checksum of the object
+        - Size (integer) - size in bytes of the object
+        - StorageClass (string) - the class of storage used to store the object
+        - Owner (dict) - the owner of the object
+        - RestoreStatus (dict) - the restoration status of the the object
+        """
         response = self.s3_client.list_objects(
             Bucket=self.bucket,
             Prefix=self.prefix,
         )
-        return [obj.get('Key') for obj in response.get('Contents', [])]
+        return response.get('Contents', [])
 
-    def get_most_recent_object(self) -> str:
-        """Return the most recent object in the given bucket/prefix."""
-        objs = self.list_objects()
-        return max(objs) if objs else None
+    def get_most_recent_object_key(self) -> str:
+        """Return the most recent object's key in the self.bucket at self.prefix."""
+        objects = self.list_objects()
+        if not objects:
+            return None
+        most_recent_object = max(objects, key=lambda x: x['LastModified'])
+        return most_recent_object['Key']
 
     def get_object_last_modified_datetime(self, object_key: str) -> datetime:
         """Get last modified datetime of a specific object."""

datahub/ingest/tasks.py

@@ -64,7 +64,7 @@ def __init__(self, prefix: str):
 
     def identify_new_objects(self, ingestion_task_function: callable) -> None:
         """Entry point method to identify new objects and, if valid, schedule their ingestion."""
-        latest_object_key = self.s3_processor.get_most_recent_object()
+        latest_object_key = self.s3_processor.get_most_recent_object_key()
 
         if not latest_object_key:
             logger.info('No objects found')

datahub/ingest/test/test_boto3.py

@@ -55,17 +55,31 @@ def test_list_objects_with_objects_in_bucket(self, s3_object_processor, test_obj
         upload_objects_to_s3(s3_object_processor, test_object_tuples)
         objects = s3_object_processor.list_objects()
         assert len(objects) == 3
-        assert all(key.startswith(TEST_PREFIX) for key in objects)
-        assert all(key.endswith('.jsonl.gz') for key in objects)
-
-    def test_get_most_recent_object(self, s3_object_processor, test_object_tuples):
-        upload_objects_to_s3(s3_object_processor, test_object_tuples)
-        most_recent_object = s3_object_processor.get_most_recent_object()
-        assert most_recent_object == f'{TEST_PREFIX}file_c.jsonl.gz'
-
-    def test_get_most_recent_object_when_empty(self, s3_object_processor):
-        most_recent_object = s3_object_processor.get_most_recent_object()
-        assert most_recent_object is None
+        assert all(obj['Key'].startswith(TEST_PREFIX) for obj in objects)
+        assert all(obj['Key'].endswith('.jsonl.gz') for obj in objects)
+
+    def test_get_most_recent_object_key(self, s3_object_processor, test_object_tuples):
+        object_content = compressed_json_faker([{'test': 'content'}])
+        object_definitions = {
+            'oldest': (f'{TEST_PREFIX}/oldest-object.json.gz', object_content),
+            'middle': (f'{TEST_PREFIX}/middle-object.json.gz', object_content),
+            'newest': (f'{TEST_PREFIX}/newest-object.json.gz', object_content),
+        }
+        monday = datetime(2024, 12, 9, 10, 0, 0, tzinfo=timezone.utc)
+        tuesday = datetime(2024, 12, 10, 10, 0, 0, tzinfo=timezone.utc)
+        wednesday = datetime(2024, 12, 11, 10, 0, 0, tzinfo=timezone.utc)
+        with freeze_time(monday) as frozen_datetime:
+            upload_objects_to_s3(s3_object_processor, [object_definitions['oldest']])
+            frozen_datetime.move_to(tuesday)
+            upload_objects_to_s3(s3_object_processor, [object_definitions['middle']])
+            frozen_datetime.move_to(wednesday)
+            upload_objects_to_s3(s3_object_processor, [object_definitions['newest']])
+        most_recent_object_key = s3_object_processor.get_most_recent_object_key()
+        assert most_recent_object_key == f'{TEST_PREFIX}/newest-object.json.gz'
+
+    def test_get_most_recent_object_key_when_empty(self, s3_object_processor):
+        most_recent_object_key = s3_object_processor.get_most_recent_object_key()
+        assert most_recent_object_key is None
 
     def test_get_object_last_modified_datetime(self, s3_object_processor):
         object_definition = (

datahub/ingest/test/test_tasks.py

@@ -145,7 +145,7 @@ def test_identify_new_objects_when_no_objects_found(
         self, identification_task, mock_scheduler, caplog,
     ):
         with (
-            mock.patch.object(S3ObjectProcessor, 'get_most_recent_object', return_value=None),
+            mock.patch.object(S3ObjectProcessor, 'get_most_recent_object_key', return_value=None),
             caplog.at_level(logging.INFO),
         ):
             identification_task.identify_new_objects(base_ingestion_task)
@@ -160,7 +160,7 @@ def test_identify_new_objects_when_job_already_queued(
     ):
         with (
             mock.patch.object(
-                S3ObjectProcessor, 'get_most_recent_object', return_value=TEST_OBJECT_KEY,
+                S3ObjectProcessor, 'get_most_recent_object_key', return_value=TEST_OBJECT_KEY,
             ),
             mock.patch.object(QueueChecker, 'is_job_queued', return_value=True),
             caplog.at_level(logging.INFO),
@@ -177,7 +177,7 @@ def test_identify_new_objects_when_job_is_running(
     ):
         with (
             mock.patch.object(
-                S3ObjectProcessor, 'get_most_recent_object', return_value=TEST_OBJECT_KEY,
+                S3ObjectProcessor, 'get_most_recent_object_key', return_value=TEST_OBJECT_KEY,
             ),
             mock.patch.object(QueueChecker, 'is_job_running', return_value=True),
             caplog.at_level(logging.INFO),
@@ -191,7 +191,7 @@ def test_identify_new_objects_when_object_already_ingested(
     ):
         with (
             mock.patch.object(
-                S3ObjectProcessor, 'get_most_recent_object', return_value=TEST_OBJECT_KEY,
+                S3ObjectProcessor, 'get_most_recent_object_key', return_value=TEST_OBJECT_KEY,
             ),
             mock.patch.object(S3ObjectProcessor, 'has_object_been_ingested', return_value=True),
             caplog.at_level(logging.INFO),
@@ -205,7 +205,7 @@ def test_identify_new_objects_schedules_ingestion_task(
     ):
         with (
             mock.patch.object(
-                S3ObjectProcessor, 'get_most_recent_object', return_value=TEST_OBJECT_KEY,
+                S3ObjectProcessor, 'get_most_recent_object_key', return_value=TEST_OBJECT_KEY,
             ),
             mock.patch.object(S3ObjectProcessor, 'has_object_been_ingested', return_value=False),
             caplog.at_level(logging.INFO),