Skip to main content

Community-maintained durable execution SDK for AWS Lambda in Python

Project description

Async Durable Execution for Python

简体中文 | 繁體中文

Build API Docs Coverage PyPI - Version PyPI - Python Version OpenSSF Scorecard License


Build reliable, long-running AWS Lambda workflows with checkpointed steps, waits, callbacks, and parallel execution.

This repository is a community-maintained fork of the original Apache-2.0 licensed AWS Durable Execution Python SDK and continues to ship under Apache License 2.0 with the upstream notices preserved.

This fork was created because the official Python SDK does not support async/await, which makes it hard to integrate cleanly with other asyncio libraries. It is specifically focused on making async Python work naturally with durable functions. The official Python SDK also lacks capabilities supported by other official SDKs, including background execution of operations and parallel execution. This SDK addresses those gaps, keeps the public API synchronous at the durable operation boundary, requires user-provided durable callables to use async def for handlers, steps, child contexts, callback submitters, and condition checks, and follows a more Pythonic style with APIs that feel natural in modern Python code.

✨ Key Features

  • Async-first durable code - Compared with the official AWS SDK, user-provided durable handlers, steps, child contexts, callback submitters, map item functions, parallel branches, and wait-for-condition checks are written with async def.
  • Background operation tasks - Durable operations such as step(...), wait(...), invoke(...), and run_in_child_context(...) return asyncio.Task objects, so independent operations can run in the background and be awaited together with asyncio.gather without using parallel() or map().
  • Simplified operation APIs - The v2 API removes config wrapper objects in favor of direct keyword arguments and clearer call sites, including keyword-only operation names.
  • Integrated local and cloud runner - Runner functionality now ships through async_durable_execution, with separate local and cloud runner factories and typed test result helpers.
  • Async Lambda client support - Install the optional aioboto extra to use an async Lambda client; otherwise the SDK uses the bundled sync client through an async adapter.
  • Replay-aware logging with stdlib logging - Standard logging loggers are enriched by durable context filtering so workflow logs remain replay safe.
  • Lambda layer packaging - The repo includes tooling and workflows to build and publish an SDK Lambda layer for functions that do not vendor dependencies directly.
  • Broader validation and docs - The project now includes expanded local/cloud runner coverage, generated API docs, coverage publishing, and updated examples for async durable workflows.

🚀 Quick Start

Install the execution SDK:

pip install async-durable-execution

For an async Lambda service client, install the optional aioboto extra:

pip install "async-durable-execution[aioboto]"

The aioboto extra installs aiobotocore, which lets the SDK create an async Lambda client for durable checkpoint and state APIs. Without it, the SDK uses the bundled botocore dependency through a threaded async adapter.

Create a durable Lambda handler:

import asyncio
import logging
from datetime import timedelta

from async_durable_execution import (
    durable_callable,
    durable_execution,
    step,
    wait,
)

logger = logging.getLogger(__name__)


@durable_callable
async def validate_order(order_id: str) -> dict:
    await asyncio.sleep(0)
    logger.info("Validating order", extra={"order_id": order_id})
    return {"order_id": order_id, "valid": True}


@durable_callable
async def create_receipt(order_id: str) -> dict:
    await asyncio.sleep(0)
    logger.info("Creating receipt", extra={"order_id": order_id})
    return {"receipt_id": f"receipt-{order_id}", "order_id": order_id}


@durable_execution
async def handler(event: dict) -> dict:
    order_id = event["order_id"]
    logger.info("Starting workflow", extra={"order_id": order_id})

    validation = await step(validate_order(order_id), name="validate_order")
    if not validation["valid"]:
        return {"status": "rejected", "order_id": order_id}

    # simulate approval (real world: use wait_for_callback)
    await wait(duration=timedelta(seconds=5), name="await_confirmation")

    receipt = await step(create_receipt(order_id), name="create_receipt")

    return {"status": "approved", "order_id": order_id, "receipt": receipt}

Async callables are required anywhere the SDK accepts user code, including map() item functions, bound parallel() branch callables, child contexts, callback submitters, and wait-for-condition checks. Those callables can be functions, instance methods, class methods, or static methods. Durable context operations are awaitable and run on the same event loop as your handler.

Durable operations return asyncio.Task objects. If you call an operation without immediately awaiting it, it is scheduled to run in the background and can be awaited later. This lets independent operations run concurrently with normal asyncio patterns:

pricing_tasks = [
    step(price_line_item(item), name=f"price-{item['sku']}")
    for item in items
]
priced_items = await asyncio.gather(*pricing_tasks)

On Python 3.12 and newer, the SDK uses asyncio.eager_task_factory so newly created operation tasks start synchronously until their first suspension point. On Python 3.10 and 3.11, eager task start is not available, so operation tasks use normal lazy asyncio task scheduling; this is only an ordering and performance difference.

Handler input is deserialized from the durable execution payload before your code runs. Empty or whitespace payloads are normalized to {}, and malformed JSON fails the invocation before user code executes.

🧪 Testing Durable Functions

The SDK includes runner helpers for testing durable functions locally or against deployed Lambda functions. The local runner executes the durable handler in process, intercepts checkpoint operations with an in-memory service client, and returns a DurableFunctionTestResult that can be inspected by operation name.

Assuming the Quick Start handler above is saved in order_workflow.py, a local test can run the same durable function:

import json

from async_durable_execution import (
    DurableFunctionTestResult,
    InvocationStatus,
    create_local_runner,
)

from order_workflow import handler


async def test_my_durable_function() -> None:
    with create_local_runner(
        handler=handler,
        input={"order_id": "order-123"},
        timeout=10,
    ) as runner:
        result: DurableFunctionTestResult = await runner.run()

    receipt = {"receipt_id": "receipt-order-123", "order_id": "order-123"}

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.result == json.dumps(
        {"status": "approved", "order_id": "order-123", "receipt": receipt}
    )

    validation_result = result.get_step("validate_order")
    assert validation_result.step_details is not None
    assert validation_result.step_details.result == json.dumps(
        {"order_id": "order-123", "valid": True}
    )

    receipt_result = result.get_step("create_receipt")
    assert receipt_result.step_details is not None
    assert receipt_result.step_details.result == json.dumps(receipt)

After deploying the same handler to Lambda, use the cloud runner to test the deployed durable function. The function name must be qualified with a version or alias, for example order-workflow:$LATEST or order-workflow:prod.

import os

from async_durable_execution import InvocationStatus, create_cloud_runner


async def test_order_workflow_in_cloud() -> None:
    with create_cloud_runner(
        function_name=os.environ["ORDER_WORKFLOW_FUNCTION_NAME"],
        region=os.environ.get("AWS_REGION", "us-east-1"),
        input={"order_id": "order-123"},
        timeout=45,
    ) as runner:
        result = await runner.run()

    receipt = {"receipt_id": "receipt-order-123", "order_id": "order-123"}

    assert result.status is InvocationStatus.SUCCEEDED
    assert result.get_deserialized_result() == {
        "status": "approved",
        "order_id": "order-123",
        "receipt": receipt,
    }

🧩 Examples

Example durable functions live in async-durable-execution-examples/src/async_durable_execution_examples/. Start with hello_world.py for the smallest complete handler.

The example tests in async-durable-execution-examples/test_examples/ are also useful as executable recipes. Browse them by operation or pattern:

  • step/, wait/, wait_for_callback/, and wait_for_condition/ for core durable operations
  • step/steps_with_gather.py for starting multiple step tasks and awaiting them together with asyncio.gather
  • map/, parallel/, and run_in_child_context/ for composition patterns
  • invoke/, with_retry/, callback/, and logger_example/ for integrations and operational behavior

For the developer workflow to run or deploy example integration tests, see the Contributing Guide.

📚 Documentation

References

💬 Feedback & Support

📄 License

See the LICENSE file for our project's licensing.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

async_durable_execution-2.0.0rc3.tar.gz (299.7 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

async_durable_execution-2.0.0rc3-py3-none-any.whl (131.2 kB view details)

Uploaded Python 3

File details

Details for the file async_durable_execution-2.0.0rc3.tar.gz.

File metadata

File hashes

Hashes for async_durable_execution-2.0.0rc3.tar.gz
Algorithm Hash digest
SHA256 670a2e4e5ebccf3861fcc329065b959e8ad17bd55e154eaac2141879f85c2c2d
MD5 fee20a1a1d05c50b17629106bbf023f0
BLAKE2b-256 5ae0477f0423a088b3ccadee81e71c201d2f778abb7684edd81ab1429e048894

See more details on using hashes here.

Provenance

The following attestation bundles were made for async_durable_execution-2.0.0rc3.tar.gz:

Publisher: pypi-publish.yml on zhongkechen/async-durable-execution

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

File details

Details for the file async_durable_execution-2.0.0rc3-py3-none-any.whl.

File metadata

File hashes

Hashes for async_durable_execution-2.0.0rc3-py3-none-any.whl
Algorithm Hash digest
SHA256 b5e14615b7c24da917b4bacfeab32fc1540cd9f4bcfe78d4544d99910f15740b
MD5 6c2bea46f3dc0a80ea981c85bef8c75e
BLAKE2b-256 745c479503749c7a761aa7d3796423a00dfd23bee1d1ecd387e4cf6eff68d9bd

See more details on using hashes here.

Provenance

The following attestation bundles were made for async_durable_execution-2.0.0rc3-py3-none-any.whl:

Publisher: pypi-publish.yml on zhongkechen/async-durable-execution

Attestations: Values shown here reflect the state when the release was signed and may no longer be current.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page