feat: complete sqlite store and function handler implementation

- Add SQLiteExecutionStore with database persistence and indexing
- Implement query system with pagination support
- Add BaseExecutionStore with shared query processing logic
- Update Executor to use new query system for efficient operations
- Complete ListDurableExecutionsByFunctionHandler with proper filtering
- Add function name validation and error handling
- Add comprehensive test coverage for all implementations
- Support concurrent access patterns with proper database handling

Author: Rares Polenciuc

src/aws_durable_execution_sdk_python_testing/execution.py

@@ -58,6 +58,29 @@ def token_sequence(self) -> int:
         """Get current token sequence value."""
         return self._token_sequence
 
+    @property
+    def status(self) -> str:
+        """Get execution status string."""
+        if not self.is_complete:
+            return "RUNNING"
+
+        if not self.result:
+            return OperationStatus.FAILED.value
+
+        match self.result.status:
+            case InvocationStatus.SUCCEEDED:
+                return OperationStatus.SUCCEEDED.value
+            case InvocationStatus.FAILED:
+                if self.result.error and self.result.error.type:
+                    error_type = self.result.error.type.lower()
+                    if "timeout" in error_type:
+                        return OperationStatus.TIMED_OUT.value
+                    if "stop" in error_type:
+                        return OperationStatus.STOPPED.value
+                return OperationStatus.FAILED.value
+            case _:
+                return OperationStatus.FAILED.value
+
     @staticmethod
     def new(input: StartDurableExecutionInput) -> Execution:  # noqa: A002
         # make a nicer arn

src/aws_durable_execution_sdk_python_testing/executor.py

@@ -11,7 +11,10 @@
     DurableExecutionInvocationOutput,
     InvocationStatus,
 )
-from aws_durable_execution_sdk_python.lambda_service import ErrorObject, OperationUpdate
+from aws_durable_execution_sdk_python.lambda_service import (
+    ErrorObject,
+    OperationUpdate,
+)
 
 from aws_durable_execution_sdk_python_testing.exceptions import (
     ExecutionAlreadyStartedException,
@@ -112,21 +115,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 = execution.status
 
         # Extract result and error from execution result
         result = None
@@ -162,99 +152,41 @@ 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 = [
+            ExecutionSummary.from_execution(execution, execution.status)
+            for execution in executions
+        ]
 
         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 +201,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/model.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from typing import Any
 
 # Import existing types from the main SDK - REUSE EVERYTHING POSSIBLE
@@ -229,6 +230,24 @@ def from_dict(cls, data: dict) -> Execution:
             end_timestamp=data.get("EndTimestamp"),
         )
 
+    @classmethod
+    def from_execution(cls, execution, status: str) -> Execution:
+        """Create ExecutionSummary from Execution object."""
+
+        execution_op = execution.get_operation_execution_started()
+        return cls(
+            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,
+        )
+
     def to_dict(self) -> dict[str, Any]:
         result = {
             "DurableExecutionArn": self.durable_execution_arn,

src/aws_durable_execution_sdk_python_testing/stores/base.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from datetime import UTC
 from enum import Enum
 from typing import TYPE_CHECKING, Protocol
 
@@ -15,6 +16,7 @@ class StoreType(Enum):
 
     MEMORY = "memory"
     FILESYSTEM = "filesystem"
+    SQLITE = "sqlite"
 
 
 class ExecutionStore(Protocol):
@@ -24,4 +26,118 @@ 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,
+        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."""
+        # Apply filters
+        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 and execution.status != status_filter:
+                continue
+
+            # Time filtering
+            if time_after or time_before:
+                try:
+                    op = execution.get_operation_execution_started()
+                    if op.start_timestamp:
+                        timestamp = (
+                            op.start_timestamp.timestamp()
+                            if hasattr(op.start_timestamp, "timestamp")
+                            else op.start_timestamp.replace(tzinfo=UTC).timestamp()
+                        )
+                        if time_after and timestamp < float(time_after):
+                            continue
+                        if time_before and timestamp > float(time_before):
+                            continue
+                except (ValueError, AttributeError):
+                    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,
+        )