feat(batch): add batch processing utility

.github/scripts/release_patch_package_json.js

@@ -18,7 +18,10 @@ if (process.argv.length < 3) {
 const basePath = resolve(process.argv[2]);
 const packageJsonPath = join(basePath, 'package.json');
 const alphaPackages = [];
-const betaPackages = ['@aws-lambda-powertools/idempotency'];
+const betaPackages = [
+  '@aws-lambda-powertools/idempotency',
+  '@aws-lambda-powertools/batch',
+];
 
 (() => {
   try {

.github/workflows/reusable-run-linting-check-and-unit-tests.yml

@@ -27,9 +27,9 @@ jobs:
         with:
           nodeVersion: ${{ matrix.version }}
       - name: Run linting
-        run: npm run lint -w packages/commons -w packages/logger -w packages/tracer -w packages/metrics -w packages/parameters -w packages/idempotency
+        run: npm run lint -w packages/commons -w packages/logger -w packages/tracer -w packages/metrics -w packages/parameters -w packages/idempotency -w packages/batch
       - name: Run unit tests
-        run: npm t -w packages/commons -w packages/logger -w packages/tracer -w packages/metrics -w packages/parameters -w packages/idempotency
+        run: npm t -w packages/commons -w packages/logger -w packages/tracer -w packages/metrics -w packages/parameters -w packages/idempotency -w packages/batch
   check-examples:
     runs-on: ubuntu-latest
     env:

README.md

@@ -36,6 +36,7 @@ You can use the library in both TypeScript and JavaScript code bases.
 * **[Metrics](https://docs.powertools.aws.dev/lambda-typescript/latest/core/metrics/)** - Custom Metrics created asynchronously via CloudWatch Embedded Metric Format (EMF)
 * **[Parameters](https://docs.powertools.aws.dev/lambda-typescript/latest/utilities/parameters/)** - High-level functions to retrieve one or more parameters from AWS SSM Parameter Store, AWS Secrets Manager, AWS AppConfig, and Amazon DynamoDB
 * **[Idempotency (beta)](https://docs.powertools.aws.dev/lambda-typescript/latest/utilities/idempotency/)** - Class method decorator, Middy middleware, and function wrapper to make your Lambda functions idempotent and prevent duplicate execution based on payload content
+* **[Batch Processing (beta)](https://docs.powertools.aws.dev/lambda-typescript/latest/utilities/batch/)** - Utility to handle partial failures when processing batches from Amazon SQS, Amazon Kinesis Data Streams, and Amazon DynamoDB Streams.
 
 ## Getting started
 

docs/snippets/batch/accessLambdaContext.ts

@@ -0,0 +1,37 @@
+import {
+  BatchProcessor,
+  EventType,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  SQSEvent,
+  SQSRecord,
+  Context,
+  SQSBatchResponse,
+} from 'aws-lambda';
+
+const processor = new BatchProcessor(EventType.SQS);
+const logger = new Logger();
+
+const recordHandler = (record: SQSRecord, lambdaContext?: Context): void => {
+  const payload = record.body;
+  if (payload) {
+    const item = JSON.parse(payload);
+    logger.info('Processed item', { item });
+  }
+  if (lambdaContext) {
+    logger.info('Remaining time', {
+      time: lambdaContext.getRemainingTimeInMillis(),
+    });
+  }
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/accessProcessedMessages.ts

@@ -0,0 +1,38 @@
+import { BatchProcessor, EventType } from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  SQSEvent,
+  SQSRecord,
+  Context,
+  SQSBatchResponse,
+} from 'aws-lambda';
+
+const processor = new BatchProcessor(EventType.SQS);
+const logger = new Logger();
+
+const recordHandler = (record: SQSRecord): void => {
+  const payload = record.body;
+  if (payload) {
+    const item = JSON.parse(payload);
+    logger.info('Processed item', { item });
+  }
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  const batch = event.Records;
+
+  processor.register(batch, recordHandler, { context });
+  const processedMessages = processor.process();
+
+  for (const message of processedMessages) {
+    const status: 'success' | 'fail' = message[0];
+    const record = message[2];
+
+    logger.info('Processed record', { status, record });
+  }
+
+  return processor.response();
+};

docs/snippets/batch/customPartialProcessor.ts

@@ -0,0 +1,98 @@
+import { randomInt } from 'node:crypto';
+import {
+  DynamoDBClient,
+  BatchWriteItemCommand,
+} from '@aws-sdk/client-dynamodb';
+import { marshall } from '@aws-sdk/util-dynamodb';
+import {
+  BasePartialProcessor,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import type {
+  SuccessResponse,
+  FailureResponse,
+  EventSourceType,
+} from '@aws-lambda-powertools/batch';
+import type { SQSEvent, Context, SQSBatchResponse } from 'aws-lambda';
+
+const tableName = process.env.TABLE_NAME || 'table-not-found';
+
+class MyPartialProcessor extends BasePartialProcessor {
+  #tableName: string;
+  #client?: DynamoDBClient;
+
+  public constructor(tableName: string) {
+    super();
+    this.#tableName = tableName;
+  }
+
+  public async asyncProcessRecord(
+    _record: EventSourceType
+  ): Promise<SuccessResponse | FailureResponse> {
+    throw new Error('Not implemented');
+  }
+
+  /**
+   * It's called once, **after** processing the batch.
+   *
+   * Here we are writing all the processed messages to DynamoDB.
+   */
+  public clean(): void {
+    // We know that the client is defined because clean() is called after prepare()
+    // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+    this.#client!.send(
+      new BatchWriteItemCommand({
+        RequestItems: {
+          [this.#tableName]: this.successMessages.map((message) => ({
+            PutRequest: {
+              Item: marshall(message),
+            },
+          })),
+        },
+      })
+    );
+  }
+
+  /**
+   * It's called once, **before** processing the batch.
+   *
+   * It initializes a new client and cleans up any existing data.
+   */
+  public prepare(): void {
+    this.#client = new DynamoDBClient({});
+    this.successMessages = [];
+  }
+
+  /**
+   * It handles how your record is processed.
+   *
+   * Here we are keeping the status of each run, `this.handler` is
+   * the function that is passed when calling `processor.register()`.
+   */
+  public processRecord(
+    record: EventSourceType
+  ): SuccessResponse | FailureResponse {
+    try {
+      const result = this.handler(record);
+
+      return this.successHandler(record, result);
+    } catch (error) {
+      return this.failureHandler(record, error as Error);
+    }
+  }
+}
+
+const processor = new MyPartialProcessor(tableName);
+
+const recordHandler = (): number => {
+  return Math.floor(randomInt(1, 10));
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/extendingFailure.ts

@@ -0,0 +1,53 @@
+import { Metrics, MetricUnits } from '@aws-lambda-powertools/metrics';
+import {
+  BatchProcessor,
+  EventType,
+  FailureResponse,
+  EventSourceType,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  SQSEvent,
+  SQSRecord,
+  Context,
+  SQSBatchResponse,
+} from 'aws-lambda';
+
+class MyProcessor extends BatchProcessor {
+  #metrics: Metrics;
+
+  public constructor(eventType: keyof typeof EventType) {
+    super(eventType);
+    this.#metrics = new Metrics({ namespace: 'test' });
+  }
+
+  public failureHandler(
+    record: EventSourceType,
+    error: Error
+  ): FailureResponse {
+    this.#metrics.addMetric('BatchRecordFailures', MetricUnits.Count, 1);
+
+    return super.failureHandler(record, error);
+  }
+}
+
+const processor = new MyProcessor(EventType.SQS);
+const logger = new Logger();
+
+const recordHandler = (record: SQSRecord): void => {
+  const payload = record.body;
+  if (payload) {
+    const item = JSON.parse(payload);
+    logger.info('Processed item', { item });
+  }
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/gettingStartedAsync.ts

@@ -0,0 +1,31 @@
+import {
+  AsyncBatchProcessor,
+  EventType,
+  asyncProcessPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import axios from 'axios'; // axios is an external dependency
+import type {
+  SQSEvent,
+  SQSRecord,
+  Context,
+  SQSBatchResponse,
+} from 'aws-lambda';
+
+const processor = new AsyncBatchProcessor(EventType.SQS);
+
+const recordHandler = async (record: SQSRecord): Promise<number> => {
+  const res = await axios.post('https://httpbin.org/anything', {
+    message: record.body,
+  });
+
+  return res.status;
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  return await asyncProcessPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/gettingStartedDynamoDBStreams.ts

@@ -0,0 +1,35 @@
+import {
+  BatchProcessor,
+  EventType,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  DynamoDBStreamEvent,
+  DynamoDBRecord,
+  Context,
+  DynamoDBBatchResponse,
+} from 'aws-lambda';
+
+const processor = new BatchProcessor(EventType.DynamoDBStreams);
+const logger = new Logger();
+
+const recordHandler = (record: DynamoDBRecord): void => {
+  if (record.dynamodb && record.dynamodb.NewImage) {
+    logger.info('Processing record', { record: record.dynamodb.NewImage });
+    const message = record.dynamodb.NewImage.Message.S;
+    if (message) {
+      const payload = JSON.parse(message);
+      logger.info('Processed item', { item: payload });
+    }
+  }
+};
+
+export const handler = async (
+  event: DynamoDBStreamEvent,
+  context: Context
+): Promise<DynamoDBBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/gettingStartedKinesis.ts

@@ -0,0 +1,30 @@
+import {
+  BatchProcessor,
+  EventType,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  KinesisStreamEvent,
+  KinesisStreamRecord,
+  Context,
+  KinesisStreamBatchResponse,
+} from 'aws-lambda';
+
+const processor = new BatchProcessor(EventType.KinesisDataStreams);
+const logger = new Logger();
+
+const recordHandler = (record: KinesisStreamRecord): void => {
+  logger.info('Processing record', { record: record.kinesis.data });
+  const payload = JSON.parse(record.kinesis.data);
+  logger.info('Processed item', { item: payload });
+};
+
+export const handler = async (
+  event: KinesisStreamEvent,
+  context: Context
+): Promise<KinesisStreamBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};

docs/snippets/batch/gettingStartedSQS.ts

@@ -0,0 +1,33 @@
+import {
+  BatchProcessor,
+  EventType,
+  processPartialResponse,
+} from '@aws-lambda-powertools/batch';
+import { Logger } from '@aws-lambda-powertools/logger';
+import type {
+  SQSEvent,
+  SQSRecord,
+  Context,
+  SQSBatchResponse,
+} from 'aws-lambda';
+
+const processor = new BatchProcessor(EventType.SQS);
+const logger = new Logger();
+
+const recordHandler = (record: SQSRecord): void => {
+  const payload = record.body;
+  if (payload) {
+    const item = JSON.parse(payload);
+    logger.info('Processed item', { item });
+  }
+};
+
+export const handler = async (
+  event: SQSEvent,
+  context: Context
+): Promise<SQSBatchResponse> => {
+  return processPartialResponse(event, recordHandler, processor, {
+    context,
+  });
+};
+export { processor };