feat: add SQLite execution store with query capabilities

- Add SQLiteExecutionStore with database persistence and indexing
- Implement ExecutionQuery system for filtering by function name, execution name, status
- Add QueryResult with pagination support (limit, offset, has_more)
- Refactor BaseExecutionStore with shared query processing logic
- Update Executor to use new query system for efficient list operations
- Add comprehensive test coverage for all store implementations
- Support concurrent access patterns with proper database handling

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/executor.py

@@ -11,7 +11,11 @@
     DurableExecutionInvocationOutput,
     InvocationStatus,
 )
-from aws_durable_execution_sdk_python.lambda_service import ErrorObject, OperationUpdate
+from aws_durable_execution_sdk_python.lambda_service import (
+    ErrorObject,
+    OperationStatus,
+    OperationUpdate,
+)
 
 from aws_durable_execution_sdk_python_testing.exceptions import (
     ExecutionAlreadyStartedException,
@@ -99,6 +103,30 @@ def get_execution(self, execution_arn: str) -> Execution:
             msg: str = f"Execution {execution_arn} not found"
             raise ResourceNotFoundException(msg) from e
 
+    @staticmethod
+    def _get_execution_status(execution: Execution) -> str:
+        """Get execution status string."""
+        if not execution.is_complete:
+            return "RUNNING"
+
+        if not execution.result:
+            return OperationStatus.FAILED.value
+
+        match execution.result.status:
+            case InvocationStatus.SUCCEEDED:
+                return OperationStatus.SUCCEEDED.value
+            case InvocationStatus.FAILED:
+                if execution.result.error and execution.result.error.type:
+                    error_type = execution.result.error.type.lower()
+                    match error_type:
+                        case t if "timeout" in t:
+                            return OperationStatus.TIMED_OUT.value
+                        case t if "stop" in t:
+                            return OperationStatus.STOPPED.value
+                return OperationStatus.FAILED.value
+            case _:
+                return OperationStatus.FAILED.value
+
     def get_execution_details(self, execution_arn: str) -> GetDurableExecutionResponse:
         """Get detailed execution information for web API response.
 
@@ -112,21 +140,8 @@ def get_execution_details(self, execution_arn: str) -> GetDurableExecutionRespon
             ResourceNotFoundException: If execution does not exist
         """
         execution = self.get_execution(execution_arn)
-
-        # Extract execution details from the first operation (EXECUTION type)
         execution_op = execution.get_operation_execution_started()
-
-        # Determine status based on execution state
-        if execution.is_complete:
-            if (
-                execution.result
-                and execution.result.status == InvocationStatus.SUCCEEDED
-            ):
-                status = "SUCCEEDED"
-            else:
-                status = "FAILED"
-        else:
-            status = "RUNNING"
+        status = self._get_execution_status(execution)
 
         # Extract result and error from execution result
         result = None
@@ -162,99 +177,56 @@ def list_executions(
         function_version: str | None = None,  # noqa: ARG002
         execution_name: str | None = None,
         status_filter: str | None = None,
-        time_after: str | None = None,  # noqa: ARG002
-        time_before: str | None = None,  # noqa: ARG002
+        time_after: str | None = None,
+        time_before: str | None = None,
         marker: str | None = None,
         max_items: int | None = None,
         reverse_order: bool = False,  # noqa: FBT001, FBT002
     ) -> ListDurableExecutionsResponse:
-        """List executions with filtering and pagination.
-
-        Args:
-            function_name: Filter by function name
-            function_version: Filter by function version
-            execution_name: Filter by execution name
-            status_filter: Filter by status (RUNNING, SUCCEEDED, FAILED)
-            time_after: Filter executions started after this time
-            time_before: Filter executions started before this time
-            marker: Pagination marker
-            max_items: Maximum items to return (default 50)
-            reverse_order: Return results in reverse chronological order
-
-        Returns:
-            ListDurableExecutionsResponse: List of executions with pagination
-        """
-        # Get all executions from store
-        all_executions = self._store.list_all()
-
-        # Apply filters
-        filtered_executions = []
-        for execution in all_executions:
-            # Filter by function name
-            if function_name and execution.start_input.function_name != function_name:
-                continue
-
-            # Filter by execution name
-            if (
-                execution_name
-                and execution.start_input.execution_name != execution_name
-            ):
-                continue
-
-            # Determine execution status
-            execution_status = "RUNNING"
-            if execution.is_complete:
-                if (
-                    execution.result
-                    and execution.result.status == InvocationStatus.SUCCEEDED
-                ):
-                    execution_status = "SUCCEEDED"
-                else:
-                    execution_status = "FAILED"
-
-            # Filter by status
-            if status_filter and execution_status != status_filter:
-                continue
-
-            # Convert to ExecutionSummary
-            execution_op = execution.get_operation_execution_started()
-            execution_summary = ExecutionSummary(
-                durable_execution_arn=execution.durable_execution_arn,
-                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.timestamp()
-                if execution_op.start_timestamp
-                else datetime.now(UTC).timestamp(),
-                end_timestamp=execution_op.end_timestamp.timestamp()
-                if execution_op.end_timestamp
-                else None,
-            )
-            filtered_executions.append(execution_summary)
-
-        # Sort by start date
-        filtered_executions.sort(key=lambda e: e.start_timestamp, reverse=reverse_order)
-
-        # Apply pagination
-        if max_items is None:
-            max_items = 50
-
-        start_index = 0
+        """List executions with filtering and pagination."""
+        # Convert marker to offset
+        offset = 0
         if marker:
             try:
-                start_index = int(marker)
+                offset = int(marker)
             except ValueError:
-                start_index = 0
+                offset = 0
 
-        end_index = start_index + max_items
-        paginated_executions = filtered_executions[start_index:end_index]
+        # Query store directly with parameters
+        executions, next_marker = self._store.query(
+            function_name=function_name,
+            execution_name=execution_name,
+            status_filter=status_filter,
+            time_after=time_after,
+            time_before=time_before,
+            limit=max_items or 50,
+            offset=offset,
+            reverse_order=reverse_order,
+        )
 
-        next_marker = None
-        if end_index < len(filtered_executions):
-            next_marker = str(end_index)
+        # Convert to ExecutionSummary objects
+        execution_summaries = []
+        for execution in executions:
+            execution_op = execution.get_operation_execution_started()
+            status = self._get_execution_status(execution)
+
+            execution_summaries.append(
+                ExecutionSummary(
+                    durable_execution_arn=execution.durable_execution_arn,
+                    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.timestamp()
+                    if execution_op.start_timestamp
+                    else datetime.now(UTC).timestamp(),
+                    end_timestamp=execution_op.end_timestamp.timestamp()
+                    if execution_op.end_timestamp
+                    else None,
+                )
+            )
 
         return ListDurableExecutionsResponse(
-            durable_executions=paginated_executions, next_marker=next_marker
+            durable_executions=execution_summaries, next_marker=next_marker
         )
 
     def list_executions_by_function(
@@ -269,22 +241,7 @@ def list_executions_by_function(
         max_items: int | None = None,
         reverse_order: bool = False,  # noqa: FBT001, FBT002
     ) -> ListDurableExecutionsByFunctionResponse:
-        """List executions for a specific function.
-
-        Args:
-            function_name: The function name to filter by
-            qualifier: Function qualifier/version
-            execution_name: Filter by execution name
-            status_filter: Filter by status (RUNNING, SUCCEEDED, FAILED)
-            time_after: Filter executions started after this time
-            time_before: Filter executions started before this time
-            marker: Pagination marker
-            max_items: Maximum items to return (default 50)
-            reverse_order: Return results in reverse chronological order
-
-        Returns:
-            ListDurableExecutionsByFunctionResponse: List of executions for the function
-        """
+        """List executions for a specific function."""
         # Use the general list_executions method with function_name filter
         list_response = self.list_executions(
             function_name=function_name,

src/aws_durable_execution_sdk_python_testing/stores/base.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from datetime import UTC
 from enum import Enum
 from typing import TYPE_CHECKING, Protocol
 
+from aws_durable_execution_sdk_python.execution import InvocationStatus
+
 
 if TYPE_CHECKING:
     from aws_durable_execution_sdk_python_testing.execution import Execution
@@ -15,6 +18,7 @@ class StoreType(Enum):
 
     MEMORY = "memory"
     FILESYSTEM = "filesystem"
+    SQLITE = "sqlite"
 
 
 class ExecutionStore(Protocol):
@@ -24,4 +28,111 @@ class ExecutionStore(Protocol):
     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
+    def query(
+        self,
+        function_name: str | None = None,
+        execution_name: str | None = None,
+        status_filter: str | None = None,
+        time_after: str | None = None,
+        time_before: str | None = None,
+        limit: int | None = None,
+        offset: int = 0,
+        reverse_order: bool = False,  # noqa: FBT001, FBT002
+    ) -> tuple[list[Execution], str | None]: ...  # pragma: no cover
+    def list_all(
+        self,
+    ) -> list[Execution]: ...  # pragma: no cover  # Keep for backward compatibility
+
+
+class BaseExecutionStore(ExecutionStore):
+    """Base implementation for execution stores with shared query logic."""
+
+    @staticmethod
+    def process_query(
+        executions: list[Execution],
+        function_name: str | None = None,
+        execution_name: str | None = None,
+        status_filter: str | None = None,
+        time_after: str | None = None,  # noqa: ARG004
+        time_before: str | None = None,  # noqa: ARG004
+        limit: int | None = None,
+        offset: int = 0,
+        reverse_order: bool = False,  # noqa: FBT001, FBT002
+    ) -> tuple[list[Execution], str | None]:
+        """Apply filtering, sorting, and pagination to executions."""
+        # Apply filters
+        # :TODO add time filtering
+        filtered = []
+        for execution in executions:
+            if function_name and execution.start_input.function_name != function_name:
+                continue
+            if (
+                execution_name
+                and execution.start_input.execution_name != execution_name
+            ):
+                continue
+
+            # Status filtering
+            if status_filter:
+                status = "RUNNING"
+                if execution.is_complete:
+                    status = (
+                        "SUCCEEDED"
+                        if execution.result
+                        and execution.result.status == InvocationStatus.SUCCEEDED
+                        else "FAILED"
+                    )
+                if status != status_filter:
+                    continue
+
+            filtered.append(execution)
+
+        # Sort by start timestamp
+        def get_sort_key(e):
+            try:
+                op = e.get_operation_execution_started()
+                if op.start_timestamp:
+                    return (
+                        op.start_timestamp.timestamp()
+                        if hasattr(op.start_timestamp, "timestamp")
+                        else op.start_timestamp.replace(tzinfo=UTC).timestamp()
+                    )
+            except Exception:  # noqa: BLE001, S110
+                pass
+            return 0
+
+        filtered.sort(key=get_sort_key, reverse=reverse_order)
+
+        # Apply pagination
+        if limit:
+            end_idx = offset + limit
+            paginated = filtered[offset:end_idx]
+            has_more = end_idx < len(filtered)
+            next_marker = str(end_idx) if has_more else None
+            return paginated, next_marker
+        return filtered[offset:], None
+
+    def query(
+        self,
+        function_name: str | None = None,
+        execution_name: str | None = None,
+        status_filter: str | None = None,
+        time_after: str | None = None,
+        time_before: str | None = None,
+        limit: int | None = None,
+        offset: int = 0,
+        reverse_order: bool = False,  # noqa: FBT001, FBT002
+    ) -> tuple[list[Execution], str | None]:
+        """Apply filtering, sorting, and pagination to executions."""
+        executions = self.list_all()
+        return self.process_query(
+            executions,
+            function_name=function_name,
+            execution_name=execution_name,
+            status_filter=status_filter,
+            time_after=time_after,
+            time_before=time_before,
+            limit=limit,
+            offset=offset,
+            reverse_order=reverse_order,
+        )

src/aws_durable_execution_sdk_python_testing/stores/filesystem.py

@@ -11,6 +11,9 @@
     ResourceNotFoundException,
 )
 from aws_durable_execution_sdk_python_testing.execution import Execution
+from aws_durable_execution_sdk_python_testing.stores.base import (
+    BaseExecutionStore,
+)
 
 
 class DateTimeEncoder(json.JSONEncoder):
@@ -35,7 +38,7 @@ def datetime_object_hook(obj):
     return obj
 
 
-class FileSystemExecutionStore:
+class FileSystemExecutionStore(BaseExecutionStore):
     """File system-based execution store for persistence."""
 
     def __init__(self, storage_dir: Path) -> None: