Skip to content

SQS worker

A Lambda triggered by a queue, with a dead-letter queue and a concurrency limit. It is built from the patterns in this book; see A complete stack.

@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},
        },
    }

Rendered YAML

cfn.to_yaml(build()), committed as sqs-worker.yaml:

AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  Stage:
    Type: String
    Default: dev
    AllowedValues:
      - dev
      - test
      - prod
Resources:
  ExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action:
              - sts:AssumeRole
      Policies:
        - PolicyName: order-processor-lambda
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - logs:CreateLogStream
                  - logs:CreateLogGroup
                  - logs:PutLogEvents
                Resource:
                  - !Sub arn:${AWS::Partition}:logs:${AWS::Region}:${AWS::AccountId}:log-group:/aws/lambda/order-processor*:*
              - Effect: Allow
                Action:
                  - sqs:ReceiveMessage
                  - sqs:ChangeMessageVisibility
                  - sqs:GetQueueUrl
                  - sqs:DeleteMessage
                  - sqs:GetQueueAttributes
                Resource:
                  - !GetAtt OrderQueue.Arn
              - Effect: Allow
                Action:
                  - dynamodb:BatchWriteItem
                  - dynamodb:PutItem
                  - dynamodb:UpdateItem
                  - dynamodb:DeleteItem
                  - dynamodb:DescribeTable
                Resource:
                  - !Sub arn:${AWS::Partition}:dynamodb:${AWS::Region}:${AWS::AccountId}:table/orders-${Stage}
  Processor:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub order-processor-${Stage}-processor
      Handler: src/processor.handler
      Runtime: nodejs20.x
      Role: !GetAtt ExecutionRole.Arn
      Code:
        S3Bucket: deploy-artifacts
        S3Key: !Sub order-processor/${Stage}/package.zip
      MemorySize: 256
      Timeout: 30
      Environment:
        Variables:
          STAGE: !Ref Stage
          TABLE_NAME: !Sub orders-${Stage}
      ReservedConcurrentExecutions: 5
      Tags:
        - Key: Service
          Value: order-processor
        - Key: Stage
          Value: !Ref Stage
  ProcessorEventSourceMapping:
    Type: AWS::Lambda::EventSourceMapping
    Properties:
      BatchSize: 10
      EventSourceArn: !GetAtt OrderQueue.Arn
      FunctionName: !GetAtt Processor.Arn
  OrderDLQ:
    Type: AWS::SQS::Queue
    Properties:
      Tags:
        - Key: Service
          Value: order-processor
        - Key: Stage
          Value: !Ref Stage
  OrderQueue:
    Type: AWS::SQS::Queue
    Properties:
      VisibilityTimeout: 180
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt OrderDLQ.Arn
        maxReceiveCount: 3
      Tags:
        - Key: Service
          Value: order-processor
        - Key: Stage
          Value: !Ref Stage
Outputs:
  QueueUrl:
    Value: !Ref OrderQueue
  DLQUrl:
    Value: !Ref OrderDLQ