diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py index 71eb01ea..f4e062b5 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/config.py @@ -14,11 +14,9 @@ P = TypeVar("P") # Payload type R = TypeVar("R") # Result type T = TypeVar("T") -U = TypeVar("U") if TYPE_CHECKING: from collections.abc import Callable - from concurrent.futures import Future from aws_durable_execution_sdk_python.lambda_service import OperationSubType from aws_durable_execution_sdk_python.retries import RetryDecision @@ -65,19 +63,6 @@ def from_days(cls, value: float) -> Duration: return cls(seconds=int(value * 86400)) -@dataclass(frozen=True) -class BatchedInput(Generic[T, U]): - batch_input: T - items: list[U] - - -class TerminationMode(Enum): - TERMINATE = "TERMINATE" - CANCEL = "CANCEL" - WAIT = "WAIT" - ABANDON = "ABANDON" - - class NestingType(Enum): """Control how child contexts are created for batch operations. @@ -308,18 +293,6 @@ class ChildConfig(Generic[T]): Applied at the handler level to serialize the entire BatchResult object. If None, uses the default JSON serializer for BatchResult. - Backward Compatibility: If only 'serdes' is provided (no item_serdes), - it will be used for both individual items AND BatchResult serialization - to maintain existing behavior. - - item_serdes: Custom serialization/deserialization configuration for individual items. - Applied to each item's result as tasks complete in child contexts. - If None, uses the default JSON serializer for individual items. - - When both 'serdes' and 'item_serdes' are provided: - - item_serdes: Used for individual item results in child contexts - - serdes: Used for the entire BatchResult at handler level - sub_type: Operation subtype identifier used for tracking and debugging. Examples: OperationSubType.MAP_ITERATION, OperationSubType.PARALLEL_BRANCH. Used internally by the execution engine for operation classification. @@ -344,58 +317,17 @@ class ChildConfig(Generic[T]): """ serdes: SerDes | None = None - item_serdes: SerDes | None = None sub_type: OperationSubType | None = None summary_generator: SummaryGenerator | None = None is_virtual: bool = False -class ItemsPerBatchUnit(Enum): - COUNT = ("COUNT",) - BYTES = "BYTES" - - -@dataclass(frozen=True) -class ItemBatcher(Generic[T]): - """Configuration for batching items in map operations. - - This class defines how individual items should be grouped together into batches - for more efficient processing in map operations. - - Args: - max_items_per_batch: Maximum number of items to include in a single batch. - If 0 (default), no item count limit is applied. Use this to control - batch size when processing many small items. - - max_item_bytes_per_batch: Maximum total size in bytes for items in a batch. - If 0 (default), no size limit is applied. Use this to control memory - usage when processing large items or when items vary significantly in size. - - batch_input: Additional data to include with each batch. - This data is passed to the processing function along with the batched items. - Useful for providing context or configuration that applies to all items - in the batch. - - Example: - # Batch up to 100 items or 1MB, whichever comes first - batcher = ItemBatcher( - max_items_per_batch=100, - max_item_bytes_per_batch=1024*1024, - batch_input={"processing_mode": "fast"} - ) - """ - - max_items_per_batch: int = 0 - max_item_bytes_per_batch: int | float = 0 - batch_input: T | None = None - - @dataclass(frozen=True) class MapConfig(Generic[T]): """Configuration options for map operations over collections. This class configures how map operations process collections of items, - including concurrency, batching, completion criteria, and serialization. + including concurrency, completion criteria, and serialization. Type Parameters: T: The type of items being processed in the map operation. @@ -405,10 +337,6 @@ class MapConfig(Generic[T]): If None, no limit is imposed and all items are processed concurrently. Use this to control resource usage when processing large collections. - item_batcher: Configuration for batching multiple items together for processing. - Allows grouping items by count or size to optimize processing efficiency. - Default is no batching (each item processed individually). - completion_config: Defines when the map operation should complete. Controls success/failure criteria for the overall map operation. Default allows any number of failures. Use CompletionConfig.all_successful() @@ -449,10 +377,9 @@ class MapConfig(Generic[T]): If None, uses the default naming: "map-item-{index}". Example: - # Process 5 items at a time, batch by count, require all to succeed + # Process 5 items at a time, require all to succeed config = MapConfig( max_concurrency=5, - item_batcher=ItemBatcher(max_items_per_batch=10), completion_config=CompletionConfig.all_successful() ) @@ -464,7 +391,6 @@ class MapConfig(Generic[T]): """ max_concurrency: int | None = None - item_batcher: ItemBatcher = field(default_factory=ItemBatcher) completion_config: CompletionConfig = field(default_factory=CompletionConfig) serdes: SerDes | None = None item_serdes: SerDes | None = None @@ -494,7 +420,6 @@ class InvokeConfig(Generic[P, R]): If provided, the invocation will be scoped to this tenant. """ - # retry_strategy: Callable[[Exception, int], RetryDecision] | None = None serdes_payload: SerDes[P] | None = None serdes_result: SerDes[R] | None = None tenant_id: str | None = None @@ -526,18 +451,6 @@ class WaitForCallbackConfig(CallbackConfig): retry_strategy: Callable[[Exception, int], RetryDecision] | None = None -class StepFuture(Generic[T]): - """A future that will block on result() until the step returns.""" - - def __init__(self, future: Future[T], name: str | None = None): - self.name = name - self.future = future - - def result(self, timeout_seconds: int | None = None) -> T: - """Return the result of the Future.""" - return self.future.result(timeout=timeout_seconds) - - # region Jitter diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py index 8c210a66..de1dc616 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/context.py @@ -16,7 +16,6 @@ ) from aws_durable_execution_sdk_python.config import ( - BatchedInput, CallbackConfig, ChildConfig, Duration, @@ -656,7 +655,7 @@ def invoke( def map( self, inputs: Sequence[U], - func: Callable[[DurableContext, U | BatchedInput[Any, U], int, Sequence[U]], T], + func: Callable[[DurableContext, U, int, Sequence[U]], T], name: str | None = None, config: MapConfig | None = None, ) -> BatchResult[R]: @@ -694,10 +693,6 @@ def map_in_child_context() -> BatchResult[R]: config=ChildConfig( sub_type=OperationSubType.MAP, serdes=getattr(config, "serdes", None), - # child_handler should only know the serdes of the parent serdes, - # the item serdes will be passed when we are actually executing - # the branch within its own child_handler. - item_serdes=None, ), ) @@ -739,10 +734,6 @@ def parallel_in_child_context() -> BatchResult[T]: config=ChildConfig( sub_type=OperationSubType.PARALLEL, serdes=getattr(config, "serdes", None), - # child_handler should only know the serdes of the parent serdes, - # the item serdes will be passed when we are actually executing - # the branch within its own child_handler. - item_serdes=None, ), ) diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py index 5db1eeb9..1db42ec3 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/lambda_service.py @@ -103,18 +103,6 @@ class CallbackTimeoutType(Enum): HEARTBEAT = "Callback.Heartbeat" -class ChainedInvokeFailedToStartType(Enum): - FAILED_TO_START = "ChainedInvoke.FailedToStart" - - -class ChainedInvokeTimeoutType(Enum): - TIMEOUT = "ChainedInvoke.Timeout" - - -class ChainedInvokeStopType(Enum): - STOPPED = "ChainedInvoke.Stopped" - - class OperationSubType(Enum): STEP = "Step" WAIT = "Wait" diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/types.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/types.py index 4fc657fb..f2e9e917 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/types.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/types.py @@ -12,7 +12,6 @@ from collections.abc import Callable, Mapping, Sequence from aws_durable_execution_sdk_python.config import ( - BatchedInput, CallbackConfig, ChildConfig, Duration, @@ -129,7 +128,7 @@ def run_in_child_context( def map( self, inputs: Sequence[U], - func: Callable[[DurableContext, U | BatchedInput[Any, U], int, Sequence[U]], T], + func: Callable[[DurableContext, U, int, Sequence[U]], T], name: str | None = None, config: MapConfig | None = None, ) -> BatchResult[T]: diff --git a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/waits.py b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/waits.py index e5aa325d..3753a1be 100644 --- a/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/waits.py +++ b/packages/aws-durable-execution-sdk-python/src/aws_durable_execution_sdk_python/waits.py @@ -27,7 +27,6 @@ class WaitStrategyConfig(Generic[T]): ) # 5 minutes backoff_rate: Numeric = 1.5 jitter_strategy: JitterStrategy = field(default=JitterStrategy.FULL) - timeout: Duration | None = None # Not implemented yet @property def initial_delay_seconds(self) -> int: @@ -39,13 +38,6 @@ def max_delay_seconds(self) -> int: """Get max delay in seconds.""" return self.max_delay.to_seconds() - @property - def timeout_seconds(self) -> int | None: - """Get timeout in seconds.""" - if self.timeout is None: - return None - return self.timeout.to_seconds() - @dataclass(frozen=True) class WaitForConditionDecision: diff --git a/packages/aws-durable-execution-sdk-python/tests/config_test.py b/packages/aws-durable-execution-sdk-python/tests/config_test.py index 17069355..4983b0ea 100644 --- a/packages/aws-durable-execution-sdk-python/tests/config_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/config_test.py @@ -1,23 +1,17 @@ """Unit tests for config module.""" -from concurrent.futures import ThreadPoolExecutor from unittest.mock import Mock from aws_durable_execution_sdk_python.config import ( - BatchedInput, CallbackConfig, ChildConfig, CompletionConfig, Duration, InvokeConfig, - ItemBatcher, - ItemsPerBatchUnit, MapConfig, ParallelConfig, StepConfig, - StepFuture, StepSemantics, - TerminationMode, ) from aws_durable_execution_sdk_python.waits import ( WaitForConditionConfig, @@ -25,13 +19,6 @@ ) -def test_batched_input(): - """Test BatchedInput dataclass.""" - batch_input = BatchedInput("batch", [1, 2, 3]) - assert batch_input.batch_input == "batch" - assert batch_input.items == [1, 2, 3] - - def test_completion_config_defaults(): """Test CompletionConfig default values.""" config = CompletionConfig() @@ -69,14 +56,6 @@ def test_completion_config_all_successful(): assert config.tolerated_failure_percentage == 0 -def test_termination_mode_enum(): - """Test TerminationMode enum.""" - assert TerminationMode.TERMINATE.value == "TERMINATE" - assert TerminationMode.CANCEL.value == "CANCEL" - assert TerminationMode.WAIT.value == "WAIT" - assert TerminationMode.ABANDON.value == "ABANDON" - - def test_parallel_config_defaults(): """Test ParallelConfig default values.""" config = ParallelConfig() @@ -183,35 +162,10 @@ def mock_summary_generator(result): assert result == "Summary of test_data" -def test_items_per_batch_unit_enum(): - """Test ItemsPerBatchUnit enum.""" - assert ItemsPerBatchUnit.COUNT.value == ("COUNT",) - assert ItemsPerBatchUnit.BYTES.value == "BYTES" - - -def test_item_batcher_defaults(): - """Test ItemBatcher default values.""" - batcher = ItemBatcher() - assert batcher.max_items_per_batch == 0 - assert batcher.max_item_bytes_per_batch == 0 - assert batcher.batch_input is None - - -def test_item_batcher_with_values(): - """Test ItemBatcher with custom values.""" - batcher = ItemBatcher( - max_items_per_batch=100, max_item_bytes_per_batch=1024, batch_input="test_input" - ) - assert batcher.max_items_per_batch == 100 - assert batcher.max_item_bytes_per_batch == 1024 - assert batcher.batch_input == "test_input" - - def test_map_config_defaults(): """Test MapConfig default values.""" config = MapConfig() assert config.max_concurrency is None - assert isinstance(config.item_batcher, ItemBatcher) assert isinstance(config.completion_config, CompletionConfig) assert config.serdes is None @@ -237,36 +191,6 @@ def test_callback_config_with_values(): assert config.serdes is serdes -def test_step_future(): - """Test StepFuture with Future.""" - with ThreadPoolExecutor(max_workers=1) as executor: - future = executor.submit(lambda: "test_result") - step_future = StepFuture(future, "test_step") - - result = step_future.result() - assert result == "test_result" - - -def test_step_future_with_timeout(): - """Test StepFuture result with timeout.""" - with ThreadPoolExecutor(max_workers=1) as executor: - future = executor.submit(lambda: "test_result") - step_future = StepFuture(future) - - result = step_future.result(timeout_seconds=1) - assert result == "test_result" - - -def test_step_future_without_name(): - """Test StepFuture without name.""" - with ThreadPoolExecutor(max_workers=1) as executor: - future = executor.submit(lambda: 42) - step_future = StepFuture(future) - - result = step_future.result() - assert result == 42 - - def test_invoke_config_defaults(): """Test InvokeConfig defaults.""" config = InvokeConfig() diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py index 8b8fe25c..212d8528 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/invoke_test.py @@ -12,7 +12,6 @@ ExecutionError, InvokeError, SuspendExecution, - TimedSuspendExecution, ) from aws_durable_execution_sdk_python.identifier import OperationIdentifier from aws_durable_execution_sdk_python.lambda_service import ( @@ -26,7 +25,6 @@ ) from aws_durable_execution_sdk_python.operation.invoke import InvokeOperationExecutor from aws_durable_execution_sdk_python.state import CheckpointedResult, ExecutionState -from aws_durable_execution_sdk_python.suspend import suspend_with_optional_resume_delay from tests.serdes_test import CustomDictSerDes @@ -392,38 +390,6 @@ def test_invoke_handler_custom_serdes_new_operation(): assert operation_update.payload == expected_serialized -def test_suspend_with_optional_resume_delay_with_timeout(): - """Test suspend_with_optional_resume_delay with timeout.""" - with pytest.raises(TimedSuspendExecution) as exc_info: - suspend_with_optional_resume_delay("test message", 30) - - assert "test message" in str(exc_info.value) - - -def test_suspend_with_optional_resume_delay_no_timeout(): - """Test suspend_with_optional_resume_delay without timeout.""" - with pytest.raises(SuspendExecution) as exc_info: - suspend_with_optional_resume_delay("test message", None) - - assert "test message" in str(exc_info.value) - - -def test_suspend_with_optional_resume_delay_zero_timeout(): - """Test suspend_with_optional_resume_delay with zero timeout.""" - with pytest.raises(SuspendExecution) as exc_info: - suspend_with_optional_resume_delay("test message", 0) - - assert "test message" in str(exc_info.value) - - -def test_suspend_with_optional_resume_delay_negative_timeout(): - """Test suspend_with_optional_resume_delay with negative timeout.""" - with pytest.raises(SuspendExecution) as exc_info: - suspend_with_optional_resume_delay("test message", -5) - - assert "test message" in str(exc_info.value) - - @pytest.mark.parametrize("status", [OperationStatus.STARTED, OperationStatus.PENDING]) def test_invoke_handler_with_operation_name(status: OperationStatus): """Test invoke_handler uses operation name in logs when available.""" diff --git a/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py b/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py index 5ebae34f..a7a6024c 100644 --- a/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/operation/map_test.py @@ -16,7 +16,6 @@ ) from aws_durable_execution_sdk_python.config import ( CompletionConfig, - ItemBatcher, MapConfig, NestingType, ) @@ -742,7 +741,6 @@ def test_map_config_with_explicit_none_summary_generator(): assert config.summary_generator is None assert config.max_concurrency is None - assert isinstance(config.item_batcher, ItemBatcher) assert isinstance(config.completion_config, CompletionConfig) assert config.serdes is None diff --git a/packages/aws-durable-execution-sdk-python/tests/suspend_test.py b/packages/aws-durable-execution-sdk-python/tests/suspend_test.py index 523ff9d4..22ab43b0 100644 --- a/packages/aws-durable-execution-sdk-python/tests/suspend_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/suspend_test.py @@ -57,9 +57,10 @@ def test_suspend_optional_timeout_with_negative(): ) -def test_suspend_optional_timeout_with_positive(): +@pytest.mark.parametrize("delay_seconds", [1, 0]) +def test_suspend_optional_timeout_with_non_negative(delay_seconds): with pytest.raises(TimedSuspendExecution, match="test"): suspend_with_optional_resume_delay( "test", - 1, + delay_seconds, ) diff --git a/packages/aws-durable-execution-sdk-python/tests/types_test.py b/packages/aws-durable-execution-sdk-python/tests/types_test.py index 7dd7e462..c9ebb8c6 100644 --- a/packages/aws-durable-execution-sdk-python/tests/types_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/types_test.py @@ -3,7 +3,6 @@ from unittest.mock import Mock from aws_durable_execution_sdk_python.config import ( - BatchedInput, CallbackConfig, ChildConfig, MapConfig, @@ -144,30 +143,6 @@ def test_callable(): mock_context.create_callback.assert_called_once_with(name=None, config=None) -def test_map_with_batched_input(): - """Test map method with BatchedInput type.""" - mock_context = Mock(spec=DurableContext) - - def map_function(ctx, item, index, items): - # item can be U or BatchedInput[Any, U] - if isinstance(item, BatchedInput): - return f"batched_{len(item.items)}" - return f"single_{item}" - - # Test with regular inputs - inputs = ["x", "y"] - mock_context.map.return_value = ["single_x", "single_y"] - result = mock_context.map(inputs, map_function) - assert result == ["single_x", "single_y"] - - # Test with BatchedInput (correct constructor) - batched_input = BatchedInput(batch_input="batch_data", items=["a", "b", "c"]) - inputs_with_batch = [batched_input] - mock_context.map.return_value = ["batched_3"] - result = mock_context.map(inputs_with_batch, map_function) - assert result == ["batched_3"] - - def test_protocol_abstract_methods(): """Test that protocol methods are abstract and contain ellipsis.""" # Test that the protocols have the expected abstract methods diff --git a/packages/aws-durable-execution-sdk-python/tests/waits_test.py b/packages/aws-durable-execution-sdk-python/tests/waits_test.py index 47c050dc..50d513d6 100644 --- a/packages/aws-durable-execution-sdk-python/tests/waits_test.py +++ b/packages/aws-durable-execution-sdk-python/tests/waits_test.py @@ -42,7 +42,6 @@ def test_default_config(self): assert config.max_delay_seconds == 300 assert config.backoff_rate == 1.5 assert config.jitter_strategy == JitterStrategy.FULL - assert config.timeout_seconds is None class TestCreateWaitStrategy: