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,7 +3,7 @@
 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 (
@@ -51,7 +51,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 +63,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 = [
+            OperationUpdate.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/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())
 

src/aws_durable_execution_sdk_python_testing/store.py

@@ -0,0 +1,29 @@
+"""Execution stores for persisting durable function executions."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol
+
+from aws_durable_execution_sdk_python_testing.stores.filesystem import (
+    FileSystemExecutionStore,
+)
+from aws_durable_execution_sdk_python_testing.stores.memory import (
+    InMemoryExecutionStore,
+)
+
+
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python_testing.execution import Execution
+
+
+class ExecutionStore(Protocol):
+    """Protocol for execution storage implementations."""
+
+    # ignore cover because coverage doesn't understand elipses
+    def save(self, execution: Execution) -> None: ...  # pragma: no cover
+    def load(self, execution_arn: str) -> Execution: ...  # pragma: no cover
+    def update(self, execution: Execution) -> None: ...  # pragma: no cover
+    def list_all(self) -> list[Execution]: ...  # pragma: no cover
+
+
+__all__ = ["ExecutionStore", "InMemoryExecutionStore", "FileSystemExecutionStore"]