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,65 @@ 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,
+        )
+
+        result = self._store.query(query)
 
-            # Determine execution status
-            execution_status = "RUNNING"
+        # Convert to ExecutionSummary objects
+        execution_summaries = []
+        for execution in result.executions:
+            execution_op = execution.get_operation_execution_started()
+            status = "RUNNING"
             if execution.is_complete:
-                if (
-                    execution.result
+                status = (
+                    "SUCCEEDED"
+                    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
+                    else "FAILED"
+                )
 
-            # 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,
+            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,
+                )
             )
-            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
-        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,13 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
+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 +19,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 +52,81 @@ 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=UTC).timestamp()
+                    )
+            except Exception:  # noqa: BLE001, S110
+                pass
+            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

@@ -5,12 +5,16 @@
 from threading import Lock
 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: