fix(sdk): allow empty operations list

src/aws_durable_execution_sdk_python/execution.py

@@ -58,10 +58,15 @@ def from_dict(input_dict: MutableMapping[str, Any]) -> InitialExecutionState:
             next_marker=input_dict.get("NextMarker", ""),
         )
 
-    def get_execution_operation(self) -> Operation:
-        if len(self.operations) < 1:
+    def get_execution_operation(self) -> Operation | None:
+        if not self.operations:
+            # Due to payload size limitations we may have an empty operations list.
+            # This will only happen when loading the initial page of results and is
+            # expected behaviour. We don't fail, but instead return None
+            # as the execution operation does not exist
             msg: str = "No durable operations found in initial execution state."
-            raise DurableExecutionsError(msg)
+            logger.debug(msg)
+            return None
 
         candidate = self.operations[0]
         if candidate.operation_type is not OperationType.EXECUTION:
@@ -71,11 +76,13 @@ def get_execution_operation(self) -> Operation:
         return candidate
 
     def get_input_payload(self) -> str | None:
-        # TODO: are these None checks necessary? i.e will there always be execution_details with input_payload
-        if execution_details := self.get_execution_operation().execution_details:
-            return execution_details.input_payload
-
-        return None
+        # It is possible that backend will not provide an execution operation
+        # for the initial page of results.
+        if not (operations := self.get_execution_operation()):
+            return None
+        if not (execution_details := operations.execution_details):
+            return None
+        return execution_details.input_payload
 
     def to_dict(self) -> MutableMapping[str, Any]:
         return {

tests/execution_test.py

@@ -788,13 +788,16 @@ def test_handler(event: Any, context: DurableContext) -> dict:
 
 
 def test_initial_execution_state_get_execution_operation_no_operations():
-    """Test get_execution_operation raises error when no operations exist."""
+    """Test get_execution_operation logs debug and returns None when no operations exist."""
     state = InitialExecutionState(operations=[], next_marker="")
 
-    with pytest.raises(
-        Exception, match="No durable operations found in initial execution state"
-    ):
-        state.get_execution_operation()
+    with patch("aws_durable_execution_sdk_python.execution.logger") as mock_logger:
+        result = state.get_execution_operation()
+
+        assert result is None
+        mock_logger.debug.assert_called_once_with(
+            "No durable operations found in initial execution state."
+        )
 
 
 def test_initial_execution_state_get_execution_operation_wrong_type():