A complete stack¶

All the patterns together: an order-processing Lambda fed by a queue, with a dead-letter queue and a concurrency cap. This is examples/sqs_worker.py.

@dataclass(frozen=True)
class WorkerConfig:
    service: str = "order-processor"
    table: str = "orders"                  # the table's name is <table>-<Stage>
    memory: int = 256
    concurrency: int = 5


def build(config: WorkerConfig = WorkerConfig()) -> cfn.Template:
    params = cfn.Params()
    stage = params.string("Stage", default="dev", allowed_values=["dev", "test", "prod"])
    tags = cfn.tags(Service=config.service, Stage=stage)
    table = f"{config.table}-${{Stage}}"          # ${Stage} is resolved by CloudFormation, in Sub

    orders = queue_with_dlq("Order", visibility_timeout=180, tags=tags)

    return {
        "AWSTemplateFormatVersion": "2010-09-09",
        "Parameters": params.declarations(),
        "Resources": cfn.combine(
            {
                "ExecutionRole": lambda_role(config.service, [
                    *grants.Sqs.Queue.consume_messages.on(orders.queue_arn),
                    *grants.DynamoDb.Table.write_data.on(cfn.arn.table(table)),
                ]),
                "Processor": lambda_.function(
                    FunctionName=cfn.sub(f"{config.service}-${{Stage}}-processor"),
                    Handler="src/processor.handler",
                    Runtime="nodejs20.x",
                    Role=cfn.get_att("ExecutionRole", "Arn"),
                    Code=lambda_.FunctionCode(S3Bucket="deploy-artifacts", S3Key=cfn.sub(f"{config.service}/${{Stage}}/package.zip")),
                    MemorySize=config.memory,
                    Timeout=30,
                    Environment=lambda_.FunctionEnvironment(Variables={"STAGE": stage, "TABLE_NAME": cfn.sub(table)}),
                    ReservedConcurrentExecutions=config.concurrency,
                    Tags=tags,
                ),
                "ProcessorEventSourceMapping": sqs_event_source("Processor", orders.queue_arn),
            },
            orders,
        ),
        "Outputs": {
            "QueueUrl": {"Value": orders.queue_url},
            "DLQUrl": {"Value": orders.dlq_url},
        },
    }

The role's permissions come from the grants (the AWS CDK's permission sets), and the other parts are the patterns from the previous pages.

See the rendered YAML.

The stage is a template parameter (Stage), so one rendered template serves every environment, and the stage is chosen at deploy time. What would change the template's shape (the table name, the memory size, the concurrency) is in WorkerConfig:

python examples/sqs_worker.py > template.json
aws cloudformation deploy --template-file template.json --stack-name order-processor-prod \
    --parameter-overrides Stage=prod --capabilities CAPABILITY_IAM

What the checker covers here¶

lambda_.function(...): every property name, type and required key.
lambda_.FunctionCode(...) and lambda_.FunctionEnvironment(...): nested shapes.
Template: the top-level structure, and that Resources holds real resources.

What it does not cover: that "OrderQueue" in cfn.get_att("OrderQueue", "Arn") names a resource that exists. Logical IDs are strings. Two habits make that safe in practice: build IDs from one name inside a pattern and expose them on a handle, as queue_with_dlq does, and run cfn-lint in CI, which does check references.

Testing it¶

The template is plain data, so a test is just a function call:

def test_dlq_is_wired() -> None:
    resources = queue_with_dlq("Order").resources
    policy = resources["OrderQueue"]["Properties"]["RedrivePolicy"]
    assert policy["deadLetterTargetArn"] == get_att("OrderDLQ", "Arn")