How to consume DynamoDB JSON and get map[string]types.AttributeValue using json.Unmarshal?

[marc-ostrow]

Confirm by changing [ ] to [x] below:

• I've gone though the API reference
• I've checked AWS Forums and StackOverflow for answers

When working with AWS SDK for Go prior to v2, I had a solution to consume a DynamoDB Streams Events. NewItem and OldItem if present are from a configurations table and an Item is defined as follows:

// Item represents a dynamoDB record in the configurations table
type Item struct {
	PK               string                 `json:"pk"`
	Version          int                    `json:"version,string"`
	ManifestVersion  string                 `json:"manifestVersion"`
	Region           string                 `json:"region"`
	K8sClusterPrefix string                 `json:"k8sClusterPrefix"`
	AZDependent      map[string]interface{} `json:"azDependent,omitempty"`
	Parameters       map[string]interface{} `json:"parameters"`
	AuthoredBy       string                 `json:"authoredBy"`
	AuthoredWhen     string                 `json:"authoredWhen"`
	ItemState        ItemState              `json:"itemState"`
	ItemStateWhen    string                 `json:"itemStateWhen"`
	ItemStateAZ      string                 `json:"itemStateAZ,omitempty"`
	Message          string                 `json:"message,omitempty"`
}

I then had the following to consume a Raw DynamoDB Streams Event

// Event represents a DynamoDB Streams Event written to SQS
type Event struct {
	Records []*EventRecord `json:"Records"`
}

// EventRecord represents a DynamoDB Streams Event Record
type EventRecord struct {
	Change    StreamRecord `json:"dynamodb"`
	EventName string       `json:"eventName"`
}

// StreamRecord represents a DynamoDB Streams Record
type StreamRecord struct {
	Keys           map[string]*dynamodb.AttributeValue `json:"Keys,omitempty"`
	NewImage       map[string]*dynamodb.AttributeValue `json:"NewImage,omitempty"`
	OldImage       map[string]*dynamodb.AttributeValue `json:"OldImage,omitempty"`
	SequenceNumber string                              `json:"SequenceNumber"`
	SizeBytes      int64                               `json:"SizeBytes"`
}

And finally, to consume messages from the SQS queue

	for {
		result, err := q.client.ReceiveMessageWithContext(q.ctx, &sqs.ReceiveMessageInput{
			QueueUrl:            sdkaws.String(env.SQSQueueURL()),
			WaitTimeSeconds:     sdkaws.Int64(maxLongPollSeconds),
			MaxNumberOfMessages: sdkaws.Int64(1),
			VisibilityTimeout:   sdkaws.Int64(visibilityTimeoutSeconds),
		})
		if err != nil {
			q.lasterr = err
			return
		}

		for _, m := range result.Messages {
			if m.Body == nil {
				q.deleteMessage(m.ReceiptHandle)
				continue
			}

			var event configurations.Event
			if err = json.Unmarshal([]byte(*m.Body), &event); err != nil {
				q.deleteMessage(m.ReceiptHandle)
				continue
			}

			for _, eventRecord := range event.Records {
				select {
				case q.eventRecordCh <- eventRecord:
				case <-q.ctx.Done():
					logrus.Info("queues.processor: context terminated")
					q.lasterr = q.ctx.Err()
					return
				}
			}

			q.deleteMessage(m.ReceiptHandle)
		}
	}

Then the consumer of the eventRecordCh channel would use

err := dynamodbattribute.UnmarshalMap(eventRecord.Change.NewImage, &item)

if a NewImage was expected; the populated map[string]*dynamodb.AttributeValue could then easily be converted to an Item.

---

When porting over to AWS SDK for go v2 the Event structures became

// Event represents a DynamoDB Streams Event written to SQS
type Event struct {
	Records []*EventRecord `json:"Records"`
}

// EventRecord represents a DynamoDB Streams Event Record
type EventRecord struct {
	Change    StreamRecord `json:"dynamodb"`
	EventName string       `json:"eventName"`
}

// StreamRecord represents a DynamoDB Streams Record
type StreamRecord struct {
	Keys           map[string]types.AttributeValue `json:"Keys,omitempty"`
	NewImage       map[string]types.AttributeValue `json:"NewImage,omitempty"`
	OldImage       map[string]types.AttributeValue `json:"OldImage,omitempty"`
	SequenceNumber string                          `json:"SequenceNumber"`
	SizeBytes      int64                           `json:"SizeBytes"`
}

And with the following test driver

	result, err := client.ReceiveMessage(context.TODO(), &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(env.SQSQueueURL()),
		WaitTimeSeconds:     20,
		MaxNumberOfMessages: 1,
		VisibilityTimeout:   30,
	})
	if err != nil {
		panic(err)
	}

	for i, m := range result.Messages {

		var event configurations.Event
		if err = json.Unmarshal([]byte(*m.Body), &event); err != nil {
			fmt.Printf("Error: %v\n", err)
			continue
		}

		fmt.Printf("result.Message[%d]: %#v\n", i, event)
	}

json.Unmarshal() returns the error

Error: json: cannot unmarshal object into Go struct field StreamRecord.Records.dynamodb.Keys of type types.AttributeValue

Mind you, I have working ported dynamodb code to read and write the table, Item changing to (yes, PK changed to a partition key 'feed' and a sort key 'microservice'; 'pk' was these two fields # concatenated)

// Item represents a dynamoDB record in the configurations table
type Item struct {
	Feed            string                 `json:"feed" dynamodbav:"feed"`
	Microservice    string                 `json:"microservice" dynamodbav:"microservice"`
	Version         int                    `json:"version,string" dynamodbav:"version,string"`
	ManifestVersion string                 `json:"manifestVersion" dynamodbav:"manifestVersion"`
	Region          string                 `json:"region" dynamodbav:"region"`
	K8sCluster      string                 `json:"k8sCluster" dynamodbav:"k8sCluster"`
	AZDependent     map[string]interface{} `json:"azDependent,omitempty" dynamodbav:"azDependent,omitempty"`
	Parameters      map[string]interface{} `json:"parameters" dynamodbav:"parameters"`
	AuthoredBy      string                 `json:"authoredBy" dynamodbav:"authoredBy"`
	AuthoredWhen    string                 `json:"authoredWhen" dynamodbav:"authoredWhen"`
	ItemState       ItemState              `json:"itemState" dynamodbav:"itemState"`
	ItemStateWhen   string                 `json:"itemStateWhen" dynamodbav:"itemStateWhen"`
	ItemStateAZ     string                 `json:"itemStateAZ,omitempty" dynamodbav:"itemStateAZ,omitempty"`
	Message         string                 `json:"message,omitempty" dynamodbav:"message,omitempty"`
}

// ItemState is string
type ItemState string

The magic that *dynamodb.AttributeValue processed allowed for json.Unmarshal() to just work. Apparently that magic wasn't passed down. ;-(

In AWS SDK for go v2, is there a public way to go from DynamoDB JSON to JSON?

[marc-ostrow]

My solution was to port the AttributeValue struct and methods from api.go in aws-sdk-go. I also threw in the Prettify internal awsutil function and helper; everything else in api.go was stripped out. I created a dynamodbattribute directory and copied the following files from aws-sdk-go as well: decode.go, encode.go, field.go, fields_go1.9.go, and tag.go; modifying them to use the parent dynamodb package and the aws package from aws-sdk-go-v2.

dynamodbevent
├── dto.go
└── dynamodb
    ├── api.go
    ├── doc.go
    └── dynamodbattribute
        ├── decode.go
        ├── encode.go
        ├── field.go
        ├── fields_go1.9.go
        └── tag.go

I restructured, moving my Event and EventRecord and StreamRecord structs to a dynamodbevent package which has the dynamodb subpackage as illustrated above.

So with

// Event represents a DynamoDB Streams Event written to SQS
type Event struct {
	Records []*EventRecord `json:"Records"`
}

// EventRecord represents a DynamoDB Streams Event Record
type EventRecord struct {
	EventName string       `json:"eventName"`
	Change    StreamRecord `json:"dynamodb"`
}

// StreamRecord represents a DynamoDB Streams Record
type StreamRecord struct {
	Keys           map[string]*dynamodb.AttributeValue `json:"Keys,omitempty"`
	NewImage       map[string]*dynamodb.AttributeValue `json:"NewImage,omitempty"`
	OldImage       map[string]*dynamodb.AttributeValue `json:"OldImage,omitempty"`
	SequenceNumber string                              `json:"SequenceNumber"`
	SizeBytes      int64                               `json:"SizeBytes"`
}

the following snippet from my test driver now works

	result, err := client.ReceiveMessage(context.TODO(), &sqs.ReceiveMessageInput{
		QueueUrl:            aws.String(env.SQSQueueURL()),
		WaitTimeSeconds:     20,
		MaxNumberOfMessages: 1,
		VisibilityTimeout:   30,
	})
	if err != nil {
		panic(err)
	}

	for _, m := range result.Messages {
		fmt.Printf("\n\nMessage Body: %v\n\n", *m.Body)

		var event dynamodbevent.Event
		if err = json.Unmarshal([]byte(*m.Body), &event); err != nil {
			fmt.Printf("Error: %v\n", err)
			continue
		}

		for _, record := range event.Records {
			if len(record.Change.NewImage) > 0 {
				var item configurations.Item
				err := dynamodbattribute.UnmarshalMap(record.Change.NewImage, &item)
				if err != nil {
					panic(err)
				}
				ij, err := json.Marshal(&item)
				if err != nil {
					panic(err)
				}
				fmt.Printf("NewImage: %v\n", string(ij))
			}
			if len(record.Change.OldImage) > 0 {
				var item configurations.Item
				err := dynamodbattribute.UnmarshalMap(record.Change.OldImage, &item)
				if err != nil {
					panic(err)
				}
				ij, err := json.Marshal(&item)
				if err != nil {
					panic(err)
				}
				fmt.Printf("OldImage: %v\n", string(ij))
			}
		}
	}

[marc-ostrow]

When aws-sdk-go-v2 supports converting DynamoDB JSON to map[string]types.AttributeValue via json.Unmarshal() I'll gladly strip the above hack out!

[milescrabill]

Check out https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue. It looks like this functionality was just moved around in v2 (out of the service dynamodb package).

[marc-ostrow]

I can successfully Scan, Query, GetItem from DynamoDB with this new version of the AWS SDK for Go. The output of such queries are
map[string]types.AttributeValue
which can then be attributevalue.UnmarshalMap'ed to an appropriate struct.

The issue is getting
map[string]types.AttributeValue from json.Unmarshal'ing DynamoDB JSON we get in DynamoDB Stream's Events.

[marc-ostrow]

The conversion of DynamoDB JSON is hidden within V2's implementation of both DynamoDB and DynamoDB Streams... I'm not reading from Shards... I have a Lamda that is called by DynamoDB Streams with an Event having one or more StreamRecords. My Lambda writes the Event to an SNS topic, which forwards to an SQS subscription. I'm reading these Events from SQS and am getting a mixture of JSON and DynamoDB JSON. At some point, I'm trying to consume NewImage and OldImage (both DynamoDB JSON) from the StreamRecord into a map[string]types.AttributeValue, which fails (using json.Unmarshal). This worked when json.Unmarshaling into map[string]*dynamodbattribute.AttributeValue in V1 of the SDK. Having the Map, of attribute values, I can use the SDK to UnmarshalMap.

[jspri]

My main use case is the same as the OPs (reading from a dynamodb stream lambda), but I ran into this issue upgrading from one of the earlier versions of the v2 SDK. It's super frustrating and my colleague spent quite a few hours pulling his hair out. I did some digging through past issues and as far as I can tell there's no reason to not support json as before.

Right now I'm trying to write a function to do

[]byte (json) -> map[string]events.DynamoDBAttributeValue (lambda go package) -> map[string]types.AttributeValue (this package) -> interface{}

and it just seems to be needlessly convoluted.

[Luke-Davies]

tl;dr: A solution is provided in the last paragraph.

I have the same use case as @marc-ostrow and @jspri of a lambda triggered by a dynamodb stream which then needs to interact with the sdk. I want to be able to use the NewImage/OldImage of the event for another PutItem call and I also want the attributevalue.UnmarshalMap convenience that the SDK provides.

I was trying to use ("github.com/aws/aws-sdk-go-v2/service/dynamodbstreams/types").Record in the event struct I defined but this has two issues - the first being the one highlighted by OP and the second being an issue that is present in the v1 SDK: ApproximateCreationDateTime cannot be unmarshalled into *time.Time because the value is unix time

There are no parse errors if you use ("github.com/aws/aws-lambda-go/events").DynamoDBEvent, which makes sense since that package is designed specifically for this use case. You then need to convert to the corresponding SDK types.

Essentially, the types in aws-sdk-go-v2 are not designed for this use case (neither were the v1 types - it was a happy coincidence that you could get them to work).

aws-lambda-go's events package was obviously designed for this use case, however neither package gives you a convenient way to convert from/to the types of the two packages and I think it's unlikely that either project will provide the utilities to do so because neither will want to be dependent on the other. Such a stance is justified by the fact that aws-lambda-go would have a decision to make on which version of the aws-sdk-go (v1 or v2... or v3 etc) to support.

The idea of aws-sdk-go-v2 providing converters for aws-lambda-go events may seem less volatile because the events structs are unlikely to undergo major version changes, however the aws-lambda-go module that the package belongs to is susceptible to a major version change in future.

In practise, this is inconvenient for anyone writing aws lambdas in Go. I think the following options should be explored:

a) aws-lambda-go should provide its own equivalent of attributevalue.UnmarshalMap - meaning we're talking about this in the wrong place.
b) aws-sdk-go-v2 should provide the events package and corresponding converters.
c) A completely separate project for converting between aws-lambda-go events and aws-sdk-go-v2 types should be created. This would only be beneficial if there are more useful type conversions required other than the one described in this issue.
d) aws-sdk-go-v2 should provide a way to unmarshal json into AttributeValue and developers should ignore aws-lambda-go's DynamoDBEvent if their lambdas need to use the sdk.

It would be great if a maintainer from each of aws-sdk-go-v2 and aws-lambda-go could provide their thoughts on how this problem should be approached.

For anyone who just wants a solution, I have adapted the code for attributevalue.FromDynamoDBStreamsMap into functions that convert from aws-lambda-go's DynamoDBEvent to aws-sdk-go-v2's dynamodbstreams []type.Record. So your lambda's event type should be events.DynamoDBEvent (from "github.com/aws/aws-lambda-go/events") and then you can use this to convert it into the sdk's dynamodbstreams []types.Record:
https://gist.github.com/Luke-Davies/eb67ddd90f73676a64ce5b0c7cf2ba80
Use at your own risk. I haven't had time to test it thoroughly.
You can adapter it to convert to the dynamodb types if required or use attributevalue.FromDynamoDBStreamsMap.

[jamessouth]

I also would like to see a SDK v2 DDB stream event type to be used in lambdas. It seems like such is missing from https://pkg.go.dev/github.com/aws/aws-sdk-go-v2/service/dynamodbstreams/types, which has:

1. a Record type that is similar to the DynamoDBEventRecord type from github.com/aws/aws-lambda-go/events

2. a StreamRecord type similar to the DynamoDBStreamRecord type from same

But there is no analogue to the DynamoDBEvent type that holds a slice of Records.

To work around I adapted Luke-Davies gist above ☝. Thank you @Luke-Davies!