Skip to content

Runner API

Import these public symbols from async_durable_execution.

Local Runner

The local runner executes durable handlers in memory without AWS credentials. Use stable operation names and inspect results by name instead of depending on operation order.

AWS Lambda Durable Executions Python SDK.

Classes

DurableFunctionLocalTestRunner 

DurableFunctionLocalTestRunner(
    handler: Callable,
    poll_interval: float = 1.0,
    input: Any = None,
    timeout: int = 900,
    function_name: str = "test-function",
    execution_name: str = "execution-name",
    account_id: str = "123456789012",
)

Methods:

mock_invoke_result 
mock_invoke_result(function_name: str, result: Any) -> None

Register a local mock result for a chained invoke function.

Functions:

create_local_runner 

create_local_runner(
    *,
    handler: Callable,
    poll_interval: float = 1.0,
    input: Any = None,
    timeout: int = 60,
) -> DurableFunctionLocalTestRunner

Create a configured local durable function runner.

Cloud Runner

The cloud runner invokes a deployed durable Lambda function qualified by version, alias, or $LATEST. See Deploy and Invoke.

AWS Lambda Durable Executions Python SDK.

Classes

DurableFunctionCloudTestRunner 

DurableFunctionCloudTestRunner(
    function_name: str,
    region: str = "us-west-2",
    lambda_endpoint: str | None = None,
    poll_interval: float = 1.0,
    input: Any = None,
    timeout: int = 60,
)

Test runner that executes durable functions against actual AWS Lambda backend.

This runner invokes deployed Lambda functions and polls for execution completion, providing the same interface as DurableFunctionLocalTestRunner for seamless test compatibility between local and cloud modes.

Initialize cloud test runner.

Methods:

close 
close() -> None

Close the underlying sync client when supported.

aclose async 
aclose() -> None

Close the underlying client when supported.

run async 
run() -> DurableFunctionTestResult

Execute function on AWS Lambda and wait for completion.

run_async async 
run_async() -> str

Execute function on AWS Lambda asynchronously

wait_for_callback async 
wait_for_callback(
    execution_arn: str,
    name: str | None = None,
    timeout: int = 60,
) -> str

Wait for and retrieve a callback ID from a durable execution.

Polls the execution history at regular intervals until a callback ID is found or the timeout is reached.

Parameters:

Name Type Description Default
execution_arn str

Execution Arn

 required
name str | None

Specific callback name, default to None

 None
timeout int

Maximum time in seconds to wait for callback. Defaults to 60.

 60

Returns:

Name Type Description
str str

The callback ID/token retrieved from the execution history

Raises:

Type Description
TimeoutError

If callback is not found within the specified timeout period
DurableFunctionsTestError

If there's an error fetching execution history (excluding retryable errors)

Functions:

create_cloud_runner 

create_cloud_runner(
    *,
    function_name: str,
    region: str = "us-west-2",
    lambda_endpoint: str | None = None,
    poll_interval: float = 1.0,
    input: Any = None,
    timeout: int = 60,
) -> DurableFunctionCloudTestRunner

Create a configured cloud durable function runner.

Test Results

Use stable operation names and inspect results by name, such as result.get_step("fetch-user"), instead of depending on operation order.

AWS Lambda Durable Executions Python SDK.

Classes

DurableFunctionTestResult dataclass 

DurableFunctionTestResult(
    status: InvocationStatus,
    operations: list[Operation],
    result: OperationPayload | None = None,
    error: ErrorObject | None = None,
    _all_operations: list[Operation] = list(),
)

Methods:

from_execution_history classmethod 
from_execution_history(
    execution_response: GetDurableExecutionResponse,
    history_response: GetDurableExecutionHistoryResponse,
) -> DurableFunctionTestResult

Create test result from execution history responses.

Factory method for cloud runner that builds DurableFunctionTestResult from GetDurableExecution and GetDurableExecutionHistory API responses.

get_deserialized_result 
get_deserialized_result(
    serdes: ExtendedTypeSerDes | None = None,
) -> Any

Return the deserialized execution result.

get_operation_deserialized_result 
get_operation_deserialized_result(
    operation: Operation,
    serdes: ExtendedTypeSerDes | None = None,
) -> Any

Return the deserialized result payload for a service operation.

get_child_operations 
get_child_operations(
    operation: Operation,
) -> list[Operation]

Return direct child operations for a service operation.

get_all_operations 
get_all_operations() -> list[Operation]

Return all non-execution operations, including nested operations.