如何将 DynamoDB Streams 对象映射到 Javascript 对象？

Question

我有一个连接到 DynamoDB Streams 的 Lambda 触发器。相关 DynamoDB 文档作为 AttributeValue 传递到 Lambda 对象。

我的 Lambda 函数如下所示：

import { Handler, Context, Callback } from 'aws-lambda'
import { config, DynamoDB } from 'aws-sdk'

const handler: Handler = async (event: any, context: Context, callback: Callback) => {
  if (!event) {
    return callback(null, `Failed processing records. event is null.`);
  }
  if (!event.Records) {
    return callback(null, `Failed processing records. No records present.`);
  }
  event.Records.forEach((record) => {
    console.log(record.eventID);
    console.log(record.eventName);
    console.log('DynamoDB Record: %j', record.dynamodb);
    //An example of record.dynamodb is shown below.
  });
  return callback(null, `Successfully processed ${event.Records.length} records.`);
}

export { handler }

AttributeValue 的示例如下所示：

{
  "ApproximateCreationDateTime": 1554660820,
  "Keys": {
    "id": {
      "S": "ab66eb3e-045e-41d6-9633-75597cd47234"
    },
    "date_time": {
      "S": "2019-04-07T18:13:40.084Z"
    }
  },
  "NewImage": {
    "some_property": {
      "S": "some data"
    },
    "another_property": {
      "S": "more data"
    },
    "id": {
      "S": "ab66eb3e-045e-41d6-9633-75597cd47234"
    },
    "date_time": {
      "S": "2019-04-07T18:13:40.084Z"
    }
  },
  "SequenceNumber": "60215400000000011976585954",
  "SizeBytes": 1693,
  "StreamViewType": "NEW_IMAGE"
}

如何将 AttributeValue 映射到 javascript 对象，最好使用 Typescript 定义？ IE。如果我像这样定义一个 Typescript 类：

class DynamoDBDocument {
    id: string;
    date_time: Date;
    some_property: string
    another_property: string
}

它会产生这样的对象：

{
    "id": "ab66eb3e-045e-41d6-9633-75597cd47234",
    "date_time": "S": "2019-04-07T18:13:40.084Z",
    "some_property": "some data",
    "another_property": "more data"
}

DynamoDB Data Mapper 看起来是一个很有前途的解决方案，但我不知道如何将它与 AttributeValue 一起使用。

Answer 1

这是使用 @shiftcoders/dynamo-easy 的解决方案：

首先定义您的模型（如果需要 ISO 字符串以外的任何内容，请确保定义正确的日期映射器）

import { DateProperty, Model, PartitionKeyUUID, SortKey } from '@shiftcoders/dynamo-easy'

@Model({ tableName: 'dynamo-table-name' })
class CustomModel {
  // hash key with auto generated uuid
  @PartitionKeyUUID()
  id: string

  // range key
  @SortKey()
  @DateProperty()
  date_time: Date

  some_property: string
  another_property: string
}

实现处理函数

import { DateProperty, fromDb, Model, PartitionKeyUUID, SortKey } from '@shiftcoders/dynamo-easy'
import { Callback, Context } from 'aws-lambda'
import * as DynamoDB from 'aws-sdk/clients/dynamodb'
import { Handler } from 'aws-sdk/clients/lambda'

const handler: Handler = async (event: any, context: Context, callback: Callback) => {
  event.Records.forEach((record) => {
    const newImage: DynamoDB.AttributeMap = record.dynamodb.NewImage
    if (newImage) {
      // map the dynamoDB attributes to a JS object using the CustomModel
      const newObject = fromDb(newImage, CustomModel)
      console.log(`item with id ${newObject.id} / ${newObject.date_time.toDateString()} changed to %j`, newObject)

      // start with the business logic which requires the newObject
    }

  })

  // no need to use callback when using async handler
  return `Successfully processed ${event.Records.length} records.`
}

查看文档以获取有关如何执行请求（如果您的业务逻辑最终需要）的更多信息。

Answer 2

我在 DynamoDB Data Marshaller 中找到了解决方案。

import { Handler, Context, Callback } from 'aws-lambda'
import { AttributeMap } from "aws-sdk/clients/dynamodb";
import { DataMapper } from '@aws/dynamodb-data-mapper'
import { unmarshallItem, Schema } from '@aws/dynamodb-data-marshaller'

const someSchema: Schema = {
    id: {type: 'String', keyType: "HASH"},
    date_time: {type: 'Date', keyType: "RANGE"},
    some_property: {type: 'String'},
    another_property: {type: 'String'},
};

const handler: Handler = async (event: any, context: Context, callback: Callback) => {

  event.Records.forEach((record) => {
    let dynamoDB: AttributeMap = record.dynamodb.NewImage
    let someObject = unmarshallItem(someSchema, dynamoDB)
    console.log('DynamoDB Object: %j', someObject)

  });
  return callback(null, `Successfully processed ${event.Records.length} records.`);
}

export { handler }

Answer 3

我发现的一种更简洁的方法是使用

aws-sdk

包中的 unmarshall 函数。它看起来像这样；

import { Converter } from 'aws-sdk/clients/dynamodb';

...

const record = Converter.unmarshall(AttributeValue); 
// where AttributeValue is the ddb stream record