feat: implement filesystem store

Author: bchampp

src/aws_durable_execution_sdk_python_testing/checkpoint/processor.py

@@ -27,7 +27,7 @@
 if TYPE_CHECKING:
     from aws_durable_execution_sdk_python_testing.execution import Execution
     from aws_durable_execution_sdk_python_testing.scheduler import Scheduler
-    from aws_durable_execution_sdk_python_testing.store import ExecutionStore
+    from aws_durable_execution_sdk_python_testing.stores import ExecutionStore
 
 
 class CheckpointProcessor:

src/aws_durable_execution_sdk_python_testing/cli.py

@@ -50,6 +50,10 @@ class CliConfig:
     local_runner_region: str = "us-west-2"
     local_runner_mode: str = "local"
 
+    # Store configuration
+    store_type: str = "memory"
+    store_path: str | None = None
+
     @classmethod
     def from_environment(cls) -> CliConfig:
         """Create configuration from environment variables with defaults."""
@@ -65,6 +69,8 @@ def from_environment(cls) -> CliConfig:
             ),
             local_runner_region=os.getenv("AWS_DEX_LOCAL_RUNNER_REGION", "us-west-2"),
             local_runner_mode=os.getenv("AWS_DEX_LOCAL_RUNNER_MODE", "local"),
+            store_type=os.getenv("AWS_DEX_STORE_TYPE", "memory"),
+            store_path=os.getenv("AWS_DEX_STORE_PATH"),
         )
 
 
@@ -175,6 +181,17 @@ def _create_start_server_parser(self, subparsers) -> None:
             default=self.config.local_runner_mode,
             help=f"Local Runner mode (default: {self.config.local_runner_mode}, env: AWS_DEX_LOCAL_RUNNER_MODE)",
         )
+        start_server_parser.add_argument(
+            "--store-type",
+            choices=["memory", "filesystem"],
+            default=self.config.store_type,
+            help=f"Store type for execution persistence (default: {self.config.store_type}, env: AWS_DEX_STORE_TYPE)",
+        )
+        start_server_parser.add_argument(
+            "--store-path",
+            default=self.config.store_path,
+            help=f"Path for filesystem store (default: {self.config.store_path or '.durable_executions'}, env: AWS_DEX_STORE_PATH)",
+        )
         start_server_parser.set_defaults(func=self.start_server_command)
 
     def _create_invoke_parser(self, subparsers) -> None:
@@ -245,6 +262,8 @@ def start_server_command(self, args: argparse.Namespace) -> int:
                 local_runner_endpoint=args.local_runner_endpoint,
                 local_runner_region=args.local_runner_region,
                 local_runner_mode=args.local_runner_mode,
+                store_type=args.store_type,
+                store_path=args.store_path,
             )
 
             logger.info(
@@ -260,6 +279,10 @@ def start_server_command(self, args: argparse.Namespace) -> int:
             logger.info("  Local Runner Endpoint: %s", args.local_runner_endpoint)
             logger.info("  Local Runner Region: %s", args.local_runner_region)
             logger.info("  Local Runner Mode: %s", args.local_runner_mode)
+            logger.info("  Store Type: %s", args.store_type)
+            if args.store_type == "filesystem":
+                store_path = args.store_path or ".durable_executions"
+                logger.info("  Store Path: %s", store_path)
 
             # Use runner as context manager for proper lifecycle
             with WebRunner(runner_config) as runner:

src/aws_durable_execution_sdk_python_testing/execution.py

@@ -3,20 +3,27 @@
 import json
 from dataclasses import replace
 from datetime import UTC, datetime
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 from uuid import uuid4
 
 from aws_durable_execution_sdk_python.execution import (
     DurableExecutionInvocationOutput,
     InvocationStatus,
 )
 from aws_durable_execution_sdk_python.lambda_service import (
+    CallbackOptions,
+    ContextOptions,
     ErrorObject,
     ExecutionDetails,
+    InvokeOptions,
     Operation,
+    OperationAction,
     OperationStatus,
+    OperationSubType,
     OperationType,
     OperationUpdate,
+    StepOptions,
+    WaitOptions,
 )
 
 # Import AWS exceptions
@@ -27,6 +34,57 @@
 from aws_durable_execution_sdk_python_testing.token import CheckpointToken
 
 
+def _operation_update_from_dict(data: dict[str, Any]) -> OperationUpdate:
+    """Create OperationUpdate from dictionary data."""
+    error = ErrorObject.from_dict(data["Error"]) if data.get("Error") else None
+
+    context_options = None
+    if context_data := data.get("ContextOptions"):
+        context_options = ContextOptions(
+            replay_children=context_data.get("ReplayChildren", False)
+        )
+
+    step_options = None
+    if step_data := data.get("StepOptions"):
+        step_options = StepOptions(
+            next_attempt_delay_seconds=step_data.get("NextAttemptDelaySeconds")
+        )
+
+    wait_options = None
+    if wait_data := data.get("WaitOptions"):
+        wait_options = WaitOptions.from_dict(wait_data)
+
+    callback_options = None
+    if callback_data := data.get("CallbackOptions"):
+        callback_options = CallbackOptions(
+            timeout_seconds=callback_data.get("TimeoutSeconds"),
+            heartbeat_timeout_seconds=callback_data.get("HeartbeatTimeoutSeconds"),
+        )
+
+    invoke_options = None
+    if invoke_data := data.get("InvokeOptions"):
+        invoke_options = InvokeOptions(
+            function_name=invoke_data.get("FunctionName", ""),
+            timeout_seconds=invoke_data.get("TimeoutSeconds"),
+        )
+
+    return OperationUpdate(
+        operation_id=data["Id"],
+        operation_type=OperationType(data["Type"]),
+        action=OperationAction(data["Action"]),
+        parent_id=data.get("ParentId"),
+        name=data.get("Name"),
+        sub_type=OperationSubType(data["SubType"]) if data.get("SubType") else None,
+        payload=data.get("Payload"),
+        error=error,
+        context_options=context_options,
+        step_options=step_options,
+        wait_options=wait_options,
+        callback_options=callback_options,
+        invoke_options=invoke_options,
+    )
+
+
 if TYPE_CHECKING:
     from aws_durable_execution_sdk_python_testing.model import (
         StartDurableExecutionInput,
@@ -51,7 +109,7 @@ def __init__(
         # TODO: this will need to persist/rehydrate depending on inmemory vs sqllite store
         self.token_sequence: int = 0
         self.is_complete: bool = False
-        self.result: DurableExecutionInvocationOutput | None
+        self.result: DurableExecutionInvocationOutput | None = None
         self.consecutive_failed_invocation_attempts: int = 0
 
     @staticmethod
@@ -63,6 +121,58 @@ def new(input: StartDurableExecutionInput) -> Execution:  # noqa: A002
             durable_execution_arn=str(uuid4()), start_input=input, operations=[]
         )
 
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize execution to dictionary."""
+        return {
+            "DurableExecutionArn": self.durable_execution_arn,
+            "StartInput": self.start_input.to_dict(),
+            "Operations": [op.to_dict() for op in self.operations],
+            "Updates": [update.to_dict() for update in self.updates],
+            "UsedTokens": list(self.used_tokens),
+            "TokenSequence": self.token_sequence,
+            "IsComplete": self.is_complete,
+            "Result": self.result.to_dict() if self.result else None,
+            "ConsecutiveFailedInvocationAttempts": self.consecutive_failed_invocation_attempts,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> Execution:
+        """Deserialize execution from dictionary."""
+        from aws_durable_execution_sdk_python_testing.model import (
+            StartDurableExecutionInput,
+        )
+
+        # Reconstruct start_input
+        start_input = StartDurableExecutionInput.from_dict(data["StartInput"])
+
+        # Reconstruct operations
+        operations = [Operation.from_dict(op_data) for op_data in data["Operations"]]
+
+        # Create execution
+        execution = cls(
+            durable_execution_arn=data["DurableExecutionArn"],
+            start_input=start_input,
+            operations=operations,
+        )
+
+        # Set additional fields
+        execution.updates = [
+            _operation_update_from_dict(update_data) for update_data in data["Updates"]
+        ]
+        execution.used_tokens = set(data["UsedTokens"])
+        execution.token_sequence = data["TokenSequence"]
+        execution.is_complete = data["IsComplete"]
+        execution.result = (
+            DurableExecutionInvocationOutput.from_dict(data["Result"])
+            if data["Result"]
+            else None
+        )
+        execution.consecutive_failed_invocation_attempts = data[
+            "ConsecutiveFailedInvocationAttempts"
+        ]
+
+        return execution
+
     def start(self) -> None:
         # not thread safe, prob should be
         if self.start_input.invocation_id is None:

src/aws_durable_execution_sdk_python_testing/executor.py

@@ -48,7 +48,7 @@
 
     from aws_durable_execution_sdk_python_testing.invoker import Invoker
     from aws_durable_execution_sdk_python_testing.scheduler import Event, Scheduler
-    from aws_durable_execution_sdk_python_testing.store import ExecutionStore
+    from aws_durable_execution_sdk_python_testing.stores import ExecutionStore
 
 logger = logging.getLogger(__name__)
 
@@ -142,15 +142,15 @@ def get_execution_details(self, execution_arn: str) -> GetDurableExecutionRespon
             durable_execution_name=execution.start_input.execution_name,
             function_arn=f"arn:aws:lambda:us-east-1:123456789012:function:{execution.start_input.function_name}",
             status=status,
-            start_timestamp=execution_op.start_timestamp.isoformat()
+            start_timestamp=execution_op.start_timestamp.timestamp()
             if execution_op.start_timestamp
-            else datetime.now(UTC).isoformat(),
+            else datetime.now(UTC).timestamp(),
             input_payload=execution_op.execution_details.input_payload
             if execution_op.execution_details
             else None,
             result=result,
             error=error,
-            end_timestamp=execution_op.end_timestamp.isoformat()
+            end_timestamp=execution_op.end_timestamp.timestamp()
             if execution_op.end_timestamp
             else None,
             version="1.0",
@@ -223,10 +223,10 @@ def list_executions(
                 durable_execution_name=execution.start_input.execution_name,
                 function_arn=f"arn:aws:lambda:us-east-1:123456789012:function:{execution.start_input.function_name}",
                 status=execution_status,
-                start_timestamp=execution_op.start_timestamp.isoformat()
+                start_timestamp=execution_op.start_timestamp.timestamp()
                 if execution_op.start_timestamp
-                else datetime.now(UTC).isoformat(),
-                end_timestamp=execution_op.end_timestamp.isoformat()
+                else datetime.now(UTC).timestamp(),
+                end_timestamp=execution_op.end_timestamp.timestamp()
                 if execution_op.end_timestamp
                 else None,
             )
@@ -333,7 +333,7 @@ def stop_execution(
         # Stop the execution
         self.fail_execution(execution_arn, stop_error)
 
-        return StopDurableExecutionResponse(stop_date=datetime.now(UTC).isoformat())
+        return StopDurableExecutionResponse(stop_date=datetime.now(UTC).timestamp())
 
     def get_execution_state(
         self,

src/aws_durable_execution_sdk_python_testing/model.py

@@ -131,11 +131,11 @@ class GetDurableExecutionResponse:
     durable_execution_name: str
     function_arn: str
     status: str
-    start_timestamp: str
+    start_timestamp: float
     input_payload: str | None = None
     result: str | None = None
     error: ErrorObject | None = None
-    end_timestamp: str | None = None
+    end_timestamp: float | None = None
     version: str | None = None
 
     @classmethod
@@ -188,8 +188,8 @@ class Execution:
     durable_execution_name: str
     function_arn: str
     status: str
-    start_timestamp: str
-    end_timestamp: str | None = None
+    start_timestamp: float
+    end_timestamp: float | None = None
 
     @classmethod
     def from_dict(cls, data: dict) -> Execution:
@@ -325,7 +325,7 @@ def to_dict(self) -> dict[str, Any]:
 class StopDurableExecutionResponse:
     """Response from stopping a durable execution."""
 
-    stop_date: str
+    stop_date: float
 
     @classmethod
     def from_dict(cls, data: dict) -> StopDurableExecutionResponse:

src/aws_durable_execution_sdk_python_testing/runner.py

@@ -48,7 +48,11 @@
     StartDurableExecutionOutput,
 )
 from aws_durable_execution_sdk_python_testing.scheduler import Scheduler
-from aws_durable_execution_sdk_python_testing.store import InMemoryExecutionStore
+from aws_durable_execution_sdk_python_testing.stores import (
+    ExecutionStore,
+    FileSystemExecutionStore,
+    InMemoryExecutionStore,
+)
 from aws_durable_execution_sdk_python_testing.web.server import WebServer
 
 
@@ -83,6 +87,10 @@ class WebRunnerConfig:
     local_runner_region: str = "us-west-2"
     local_runner_mode: str = "local"
 
+    # Store configuration
+    store_type: str = "memory"  # "memory" or "filesystem"
+    store_path: str | None = None  # Path for filesystem store
+
 
 @dataclass(frozen=True)
 class Operation:
@@ -543,7 +551,7 @@ def __init__(self, config: WebRunnerConfig) -> None:
         self._config = config
         self._server: WebServer | None = None
         self._scheduler: Scheduler | None = None
-        self._store: InMemoryExecutionStore | None = None
+        self._store: ExecutionStore | None = None
         self._invoker: LambdaInvoker | None = None
         self._executor: Executor | None = None
 
@@ -581,7 +589,11 @@ def start(self) -> None:
             raise DurableFunctionsLocalRunnerError(msg)
 
         # Create dependencies and server
-        self._store = InMemoryExecutionStore()
+        if self._config.store_type == "filesystem":
+            store_path = self._config.store_path or ".durable_executions"
+            self._store = FileSystemExecutionStore(store_path)
+        else:
+            self._store = InMemoryExecutionStore()
         self._scheduler = Scheduler()
         self._invoker = LambdaInvoker(self._create_boto3_client())