Benzene

Getting started: Benzene on AWS Lambda

Take a set of transport-neutral Benzene handlers and host them on AWS Lambda — reached over API Gateway (HTTP), SQS, and SNS, all in one function — and publish events back out over SNS. One function, one pipeline, the handlers unchanged. Adding a source is a line of host wiring, never a change to your logic.

This guide goes from pip install benzene-aws to a Lambda handler that answers POST /orders, consumes the orders:created event over both SQS and SNS, and republishes it — all exercised in-memory with no AWS account. It builds on the base tutorial: read Getting started first for the handler / Result / @message fundamentals; here we only add the AWS host.

Runnable version: examples/aws_orders is exactly this guide — the shared orders_domain hosted on Lambda, with dogfooded tests that push a native API Gateway, SQS, and SNS event through the real bindings. Read it alongside this page.

Prerequisites

1. Install the package

pip install benzene-aws            # add [boto3] for the real outbound clients

The distribution is benzene-aws. It depends on benzene-core (the pipeline and message handlers) and benzene-http (the API Gateway binding reuses the HTTP router), so a single install pulls in everything the inbound bindings and the in-memory test host need. boto3 is only required by the real SNS/SQS outbound clients and is an optional extra:

pip install "benzene-aws[boto3]"   # when you publish events from Lambda

The whole thing is importable from one module:

from benzene.aws import AwsLambdaApp, SnsMessageSender, SqsMessageSender, to_lambda_handler

2. Write handlers (transport-neutral)

Business logic lives in plain async handlers that never see AWS — the same handlers you'd host over a standalone HTTP server, GCP, or Azure. In the example they live in the shared orders_domain package. The shape is the one from Getting started: a factory closes over the handler's collaborators (an order store, an outbound MessageSender) and returns the async function.

# orders_domain/handlers.py  (excerpt)
from benzene.core import Handler, MessageSender
from benzene.results import Result

from .model import ORDER_CREATED_TOPIC, OrderCreated, PlaceOrder


def make_place_order(service: OrderService, sender: MessageSender) -> Handler:
    async def place_order(request: PlaceOrder) -> Result:
        if not request.sku:
            return Result.bad_request("sku is required")
        order = service.place(request.sku, request.quantity)
        await sender.send_message(ORDER_CREATED_TOPIC, OrderCreated(id=order.id, sku=order.sku))
        return Result.created(order)          # ingress -> handler -> egress

    return place_order


def make_on_order_created(seen: list[str]) -> Handler:
    async def on_order_created(request: OrderCreated) -> Result:
        seen.append(request.id)               # the subscriber side
        return Result.ok()

    return on_order_created

These handlers are wired onto an HttpRouter (for the routes) and a Registry (all topics, including the pub/sub subscriber) inside a single composition root — a BenzeneStartUp subclass, OrdersStartUp, that both deployment and tests boot from. The orders:created subscriber is registered as a topic, so the same handler answers the event whether it arrives over SQS or SNS. Nothing in orders_domain mentions Lambda.

3. Build the AWS host

Only one file is AWS-specific. It boots the shared OrdersStartUp, overrides the single outbound edge with a real SnsMessageSender, and specializes the app to Lambda with AwsLambdaApp:

# aws_orders/host.py
import os

from benzene.aws import AwsLambdaApp, SnsMessageSender
from benzene.core import Container, MessageSender, build_application
from orders_domain import OrdersStartUp


def build_aws_orders_app() -> AwsLambdaApp:
    topic_arn = os.environ.get("BENZENE_SNS_TOPIC_ARN")
    if not topic_arn:
        raise RuntimeError(
            "Set BENZENE_SNS_TOPIC_ARN to run the AWS host (tests use create_test_host instead)."
        )

    def use_sns(services: Container) -> None:
        services.add_instance(MessageSender, SnsMessageSender(topic_arn))

    definition, _ = build_application(OrdersStartUp, overrides=[use_sns])
    return AwsLambdaApp.from_definition(definition)

Two things to notice:

The composition root is shared with every other host; only the outbound MessageSender differs between deployment (real SNS) and tests (a fake). That single seam is what makes the tests in step 5 possible.

4. Wire the Lambda entry point

The handler AWS invokes is produced by to_lambda_handler — it wraps the app in the handler(event, context) callable Lambda expects:

# aws_orders/main.py
from benzene.aws import to_lambda_handler

from .host import build_aws_orders_app

handler = to_lambda_handler(build_aws_orders_app())

Point your Lambda's handler string at this attribute — main.handler (or aws_orders.main.handler if you package the module inside a package). That single callable dispatches by event shape (transport-bindings):

Classification happens in benzene.aws.event_source(event); an event that is none of the three raises ValueError.

5. Test every source in memory (dogfooded)

Before deploying, drive the real bindings in-memory with create_test_host(...).build_aws(). It boots your actual OrdersStartUp — the same construction the deployed handler performs — and returns an AwsLambdaTestHost you push native events into. Fake only the external edge (the outbound client); everything else is the real pipeline, routing, and handlers.

# aws_orders/tests/test_aws_orders.py
import json

from benzene.aws.testing import SqsEventBuilder
from benzene.core import MessageSender
from benzene.testing import FakeMessageSender, create_test_host
from orders_domain import ORDER_CREATED_TOPIC, OrderEventLog, OrderService, OrdersStartUp


def make_host():
    service = OrderService()
    sender = FakeMessageSender()
    seen: list[str] = []

    def overrides(services):
        services.add_instance(OrderService, service)
        services.add_instance(MessageSender, sender)     # only the external edge is faked
        services.add_instance(OrderEventLog, seen)

    host = create_test_host(OrdersStartUp).with_services(overrides).build_aws()
    return host, service, sender, seen

FakeMessageSender records what was published instead of calling AWS, so a test can assert that ingress reached egress. .build_aws() is the only AWS-specific line — swap it for .build_gcp() or .build_azure() and the same test runs against another cloud.

API Gateway ingress → handler → SNS egress:

def test_api_gateway_place_order_creates_and_publishes():
    host, service, sender, _ = make_host()

    response = host.send_http("POST", "/orders", body={"sku": "ABC", "quantity": 2})

    assert response.status_code == 201
    order = json.loads(response.body)
    assert sender.last_topic == ORDER_CREATED_TOPIC      # the handler published on the way out
    assert sender.last_message.id == order["id"]
    assert order["id"] in service.orders

send_http returns an ApiGatewayResponse (.status_code, .headers, .body) — the Benzene status created mapped to HTTP 201.

SQS and SNS ingress reach the same subscriber:

def test_sqs_order_created_is_handled():
    host, _, _, seen = make_host()
    result = host.send_sqs(ORDER_CREATED_TOPIC, {"id": "ord-sqs", "sku": "ABC"})
    assert result.batch_item_failures == []              # SQS partial-batch protocol
    assert seen == ["ord-sqs"]


def test_sns_order_created_is_handled():
    host, _, _, seen = make_host()
    host.send_sns(ORDER_CREATED_TOPIC, {"id": "ord-sns", "sku": "ABC"})   # fire-and-forget, no return
    assert seen == ["ord-sns"]

send_sqs / send_sqs_event return an SqsBatchResponse — assert on .batch_item_failures (or .item_identifiers), the object mirror of the SQS partial-batch protocol. send_sns returns nothing, matching SNS's fire-and-forget delivery.

SQS partial-batch failure — only the bad record is reported, the good one still processes:

def test_sqs_partial_batch_failure_reports_only_failed_record():
    host, _, _, seen = make_host()
    event = (
        SqsEventBuilder()
        .with_message(ORDER_CREATED_TOPIC, {"id": "ok-1", "sku": "A"}, message_id="m1")
        .with_message("orders:unknown", {}, message_id="m2")   # no handler -> not-found -> fails
        .build()
    )
    result = host.send_sqs_event(event)
    assert result.batch_item_failures == [{"itemIdentifier": "m2"}]
    assert seen == ["ok-1"]                                     # the good record still processed

SqsEventBuilder (and SnsEventBuilder / ApiGatewayRequestBuilder) build the exact native event shapes AWS delivers, so these tests exercise the real decoders, not a mock of them. Run them with no cloud:

pytest examples/aws_orders

See the testing reference and benzene.aws testing for the full surface.

6. Deploy (sketch)

Package the module together with its benzene-* dependencies (add benzene-aws[boto3] to your requirements.txt so the SNS client's boto3 is present in the deployment bundle), then:

  1. Create the Lambda function and set its handler to main.handler (or your packaged path).
  2. Set the environment variable BENZENE_SNS_TOPIC_ARN to the ARN of the SNS topic the place_order handler publishes to.
  3. Attach the triggers — all three flow into the same function:
    • an API Gateway proxy integration (REST v1 or HTTP API v2 both work — the binding detects either shape),
    • an SQS event-source mapping (enable ReportBatchItemFailures so batchItemFailures is honored),
    • an SNS subscription.

There is no framework-specific deployment tooling — this is an ordinary Python Lambda, so package it however you already do (a zip, container image, SAM, CDK, Terraform, or the console). The Hosting on AWS Lambda cookbook has the compact end-to-end recap.

7. Supported event sources

benzene.aws binds three Lambda event sources, all through the one function:

Source Topic comes from Response On handler failure
API Gateway the route (path → topic, via benzene.http) API Gateway proxy response; Benzene status → HTTP code error status → HTTP error code
SQS the topic message attribute {"batchItemFailures": [...]} that record's id reported for redelivery
SNS the topic message attribute none (fire-and-forget) raises → Lambda retries the invocation

The topic attribute for SQS/SNS is written automatically by any Benzene outbound client (SnsMessageSender / SqsMessageSender), so a Benzene-to-Benzene flow needs no extra configuration. For SQS and SNS the message body is the serialized payload and message attributes become Benzene headers (benzene.core.read_message_metadata).

Compared with the .NET port: the Python benzene.aws package currently binds API Gateway, SQS, and SNS only. The .NET host additionally offers EventBridge, DynamoDB Streams, Kafka/MSK, S3, and Kinesis event sources, plus a UsePresetTopic option for raw (non-Benzene) SQS producers. Those are not yet available in Python — don't reach for them here.

8. IAM / permissions

The example needs exactly these permissions, driven by the code you saw above:

API Gateway similarly invokes the function via a resource-based permission and needs no execution-role IAM to receive requests.

9. Observability

Compared with the .NET port: Python's benzene.aws does not yet ship the .NET host's invocation feature (UseBenzeneInvocation), automatic Activity/OpenTelemetry tracing, W3C-trace-context middleware, or log-enrichment middleware. Correlation is available at the header level as described above; deeper instrumentation is on the .NET port only for now.

10. Troubleshooting

See also