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."""
@@ -69,6 +73,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"),
         )
 
 
@@ -188,6 +194,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:
@@ -258,6 +275,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(
@@ -273,6 +292,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/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())
 

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"]

src/aws_durable_execution_sdk_python_testing/stores/__init__.py

@@ -0,0 +1,101 @@
+"""File system-based execution store implementation."""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+import threading
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from aws_durable_execution_sdk_python_testing.execution import Execution
+
+
+class DateTimeEncoder(json.JSONEncoder):
+    """Custom JSON encoder that handles datetime objects."""
+
+    def default(self, obj):
+        if isinstance(obj, datetime):
+            return obj.timestamp()
+        return super().default(obj)
+
+
+def datetime_object_hook(obj):
+    """JSON object hook to convert unix timestamps back to datetime objects."""
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if isinstance(value, int | float) and key.endswith(("_timestamp", "_time")):
+                with contextlib.suppress(ValueError, OSError):
+                    obj[key] = datetime.fromtimestamp(value, tz=UTC)
+    return obj
+
+
+class FileSystemExecutionStore:
+    """File system-based execution store for persistence."""
+
+    def __init__(self, storage_dir: str | Path = ".durable_executions") -> None:
+        self._storage_dir = Path(storage_dir)
+        self._storage_dir.mkdir(exist_ok=True)
+        self._lock = threading.RLock()
+
+    def _get_file_path(self, execution_arn: str) -> Path:
+        """Get file path for execution ARN."""
+        # Use ARN as filename with .json extension, replacing unsafe characters
+        safe_filename = execution_arn.replace(":", "_").replace("/", "_")
+        return self._storage_dir / f"{safe_filename}.json"
+
+    def save(self, execution: Execution) -> None:
+        """Save execution to file system."""
+        with self._lock:
+            file_path = self._get_file_path(execution.durable_execution_arn)
+            data = execution.to_dict()
+
+            # Write atomically using temporary file
+            temp_path = file_path.with_suffix(".tmp")
+            try:
+                with open(temp_path, "w", encoding="utf-8") as f:
+                    json.dump(data, f, indent=2, cls=DateTimeEncoder)
+                temp_path.replace(file_path)
+            except Exception:
+                if temp_path.exists():
+                    temp_path.unlink()
+                raise
+
+    def load(self, execution_arn: str) -> Execution:
+        """Load execution from file system."""
+        from aws_durable_execution_sdk_python_testing.execution import Execution
+
+        with self._lock:
+            file_path = self._get_file_path(execution_arn)
+            if not file_path.exists():
+                msg = f"Execution {execution_arn} not found"
+                raise KeyError(msg)
+
+            with open(file_path, encoding="utf-8") as f:
+                data = json.load(f, object_hook=datetime_object_hook)
+
+            return Execution.from_dict(data)
+
+    def update(self, execution: Execution) -> None:
+        """Update execution in file system (same as save)."""
+        self.save(execution)
+
+    def list_all(self) -> list[Execution]:
+        """List all executions from file system."""
+        from aws_durable_execution_sdk_python_testing.execution import Execution
+
+        with self._lock:
+            executions = []
+            for file_path in self._storage_dir.glob("*.json"):
+                try:
+                    with open(file_path, encoding="utf-8") as f:
+                        data = json.load(f, object_hook=datetime_object_hook)
+                    executions.append(Execution.from_dict(data))
+                except (json.JSONDecodeError, KeyError, OSError) as e:
+                    logging.warning("Skipping corrupted file %s: %s", file_path, e)
+                    continue
+            return executions