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

@@ -162,8 +162,8 @@ 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
@@ -184,77 +184,53 @@ def list_executions(
         Returns:
             ListDurableExecutionsResponse: List of executions with pagination
         """
-        # Get all executions from store
-        all_executions = self._store.list_all()
+        from aws_durable_execution_sdk_python_testing.stores.base import ExecutionQuery
 
-        # 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
+        # Convert marker to offset
+        offset = 0
+        if marker:
+            try:
+                offset = int(marker)
+            except ValueError:
+                offset = 0
 
-            # Filter by execution name
-            if (
-                execution_name
-                and execution.start_input.execution_name != execution_name
-            ):
-                continue
+        # Query store
+        query = ExecutionQuery(
+            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
+        )
 
-            # 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
+        result = self._store.query(query)
+
+        # Convert to ExecutionSummary objects
+        execution_summaries = []
+        for execution in result.executions:
             execution_op = execution.get_operation_execution_started()
-            execution_summary = ExecutionSummary(
+            status = "RUNNING"
+            if execution.is_complete:
+                status = "SUCCEEDED" if execution.result and execution.result.status == InvocationStatus.SUCCEEDED else "FAILED"
+
+            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=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)
+                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,
+            ))
 
-        # Apply pagination
-        if max_items is None:
-            max_items = 50
-
-        start_index = 0
-        if marker:
-            try:
-                start_index = int(marker)
-            except ValueError:
-                start_index = 0
-
-        end_index = start_index + max_items
-        paginated_executions = filtered_executions[start_index:end_index]
-
-        next_marker = None
-        if end_index < len(filtered_executions):
-            next_marker = str(end_index)
+        # Generate next marker
+        next_marker = str(offset + len(execution_summaries)) if result.has_more 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(

src/aws_durable_execution_sdk_python_testing/stores/base.py

@@ -2,9 +2,12 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+from datetime import timezone
 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,30 @@ class StoreType(Enum):
 
     MEMORY = "memory"
     FILESYSTEM = "filesystem"
+    SQLITE = "sqlite"
+
+
+@dataclass
+class ExecutionQuery:
+    """Query parameters for execution filtering."""
+    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
+
+
+@dataclass
+class QueryResult:
+    """Result of a query with pagination info."""
+    executions: list[Execution]
+    total_count: int
+    has_more: bool
+
+
 
 
 class ExecutionStore(Protocol):
@@ -24,4 +51,64 @@ 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, query: ExecutionQuery) -> QueryResult: ...  # 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], query: ExecutionQuery) -> QueryResult:
+        """Apply filtering, sorting, and pagination to executions."""
+        # Apply filters
+        filtered = []
+        for execution in executions:
+            if query.function_name and execution.start_input.function_name != query.function_name:
+                continue
+            if query.execution_name and execution.start_input.execution_name != query.execution_name:
+                continue
+
+            # Status filtering
+            if query.status_filter:
+                status = "RUNNING"
+                if execution.is_complete:
+                    status = "SUCCEEDED" if execution.result and execution.result.status == InvocationStatus.SUCCEEDED else "FAILED"
+                if status != query.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=timezone.utc).timestamp()
+                else:
+                    return 0
+            except Exception:  # noqa: BLE001
+                return 0
+
+        filtered.sort(key=get_sort_key, reverse=query.reverse_order)
+
+        total_count = len(filtered)
+
+        # Apply pagination
+        if query.limit:
+            end_idx = query.offset + query.limit
+            paginated = filtered[query.offset:end_idx]
+            has_more = end_idx < total_count
+        else:
+            paginated = filtered[query.offset:]
+            has_more = False
+
+        return QueryResult(
+            executions=paginated,
+            total_count=total_count,
+            has_more=has_more
+        )
+
+    def query(self, query: ExecutionQuery) -> QueryResult:
+        """Apply filtering, sorting, and pagination to executions."""
+        executions = self.list_all()
+        return self.process_query(executions, query)

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:

src/aws_durable_execution_sdk_python_testing/stores/memory.py

@@ -6,11 +6,17 @@
 from typing import TYPE_CHECKING
 
 
+
+from aws_durable_execution_sdk_python_testing.stores.base import (
+    BaseExecutionStore,
+)
+
+
 if TYPE_CHECKING:
     from aws_durable_execution_sdk_python_testing.execution import Execution
 
 
-class InMemoryExecutionStore:
+class InMemoryExecutionStore(BaseExecutionStore):
     """Dict-based storage for testing."""
 
     def __init__(self) -> None: