Fix bug in S3KeysUnchangedTrigger and is_key_unchanged in hook

astronomer/providers/amazon/aws/hooks/s3.py

@@ -215,7 +215,7 @@ async def is_keys_unchanged(
         inactivity_seconds: int = 0,
         allow_delete: bool = True,
         last_activity_time: Optional[datetime] = None,
-    ) -> Dict[str, str]:
+    ) -> Dict[str, Any]:
         """
         Checks whether new objects have been uploaded and the inactivity_period
         has passed and updates the state of the sensor accordingly.
@@ -252,7 +252,12 @@ async def is_keys_unchanged(
             last_activity_time = datetime.now()
             inactivity_seconds = 0
             previous_objects = current_objects
-            return {"status": "pending"}
+            return {
+                "status": "pending",
+                "previous_objects": previous_objects,
+                "last_activity_time": last_activity_time,
+                "inactivity_seconds": inactivity_seconds,
+            }
 
         if len(previous_objects) - len(current_objects):
             # During the last poke interval objects were deleted.
@@ -265,7 +270,12 @@ async def is_keys_unchanged(
                     "file counter and resetting last_activity_time:\n%s",
                     deleted_objects,
                 )
-                return {"status": "pending"}
+                return {
+                    "status": "pending",
+                    "previous_objects": previous_objects,
+                    "last_activity_time": last_activity_time,
+                    "inactivity_seconds": inactivity_seconds,
+                }
 
             return {
                 "status": "error",
@@ -298,4 +308,9 @@ async def is_keys_unchanged(
                 "status": "error",
                 "message": f"FAILURE: Inactivity Period passed, not enough objects found in {path}",
             }
-        return {"status": "pending"}
+        return {
+            "status": "pending",
+            "previous_objects": previous_objects,
+            "last_activity_time": last_activity_time,
+            "inactivity_seconds": inactivity_seconds,
+        }

astronomer/providers/amazon/aws/triggers/s3.py

@@ -129,7 +129,7 @@ def __init__(
         inactivity_period: float = 60 * 60,
         min_objects: int = 1,
         inactivity_seconds: int = 0,
-        previous_objects: Optional[Set[str]] = None,
+        previous_objects: Optional[Set[str]] = set(),
         allow_delete: bool = True,
         aws_conn_id: str = "aws_default",
         last_activity_time: Optional[datetime] = None,
@@ -184,9 +184,13 @@ async def run(self) -> AsyncIterator["TriggerEvent"]:  # type: ignore[override]
                         self.allow_delete,
                         self.last_activity_time,
                     )
-                    if result.get("status") == "success" or result.get("error") == "error":
+                    if result.get("status") == "success" or result.get("status") == "error":
                         yield TriggerEvent(result)
                         return
+                    elif result.get("status") == "pending":
+                        self.previous_objects = result.get("previous_objects", set())
+                        self.last_activity_time = result.get("last_activity_time")
+                        self.inactivity_seconds = result.get("inactivity_seconds", 0)
         except Exception as e:
             yield TriggerEvent({"status": "error", "message": str(e)})
 

tests/amazon/aws/hooks/test_s3_hooks.py

@@ -280,7 +280,7 @@ async def test_s3_key_hook_is_keys_unchanged_false(mock_list_keys, mock_client):
         last_activity_time=None,
     )
 
-    assert response == {"status": "pending"}
+    assert response.get("status") == "pending"
 
     # test for the case when current_objects < previous_objects
     mock_list_keys.return_value = []
@@ -298,7 +298,7 @@ async def test_s3_key_hook_is_keys_unchanged_false(mock_list_keys, mock_client):
         last_activity_time=None,
     )
 
-    assert response == {"status": "pending"}
+    assert response.get("status") == "pending"
 
 
 @mock.patch("astronomer.providers.amazon.aws.triggers.s3.S3HookAsync.get_client_async")