-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
11 changed files
with
549 additions
and
85 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| package config | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/artie-labs/reader/lib/s3lib" | ||
| "github.com/artie-labs/transfer/lib/stringutil" | ||
| ) | ||
|
|
||
| type DynamoDB struct { | ||
| OffsetFile string `yaml:"offsetFile"` | ||
| AwsRegion string `yaml:"awsRegion"` | ||
| AwsAccessKeyID string `yaml:"awsAccessKeyId"` | ||
| AwsSecretAccessKey string `yaml:"awsSecretAccessKey"` | ||
| StreamArn string `yaml:"streamArn"` | ||
| TableName string `yaml:"tableName"` | ||
|
|
||
| Snapshot bool `yaml:"snapshot"` | ||
| SnapshotSettings *SnapshotSettings `yaml:"snapshotSettings"` | ||
| } | ||
|
|
||
| func (d *DynamoDB) Validate() error { | ||
| if d == nil { | ||
| return fmt.Errorf("dynamodb config is nil") | ||
| } | ||
|
|
||
| if stringutil.Empty(d.OffsetFile, d.AwsRegion, d.AwsAccessKeyID, d.AwsSecretAccessKey, d.StreamArn, d.TableName) { | ||
| return fmt.Errorf("one of the dynamoDB configs is empty: offsetFile, awsRegion, awsAccessKeyID, awsSecretAccessKey, streamArn or tableName") | ||
| } | ||
|
|
||
| if d.Snapshot { | ||
| if err := d.SnapshotSettings.Validate(); err != nil { | ||
| return fmt.Errorf("snapshot validation failed - err: %v", err) | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| type SnapshotSettings struct { | ||
| Folder string `yaml:"folder"` | ||
| // If the files are not specified, that's okay. | ||
| // We will scan the folder and then load into `specifiedFiles` | ||
| SpecifiedFiles []s3lib.S3File `yaml:"specifiedFiles"` | ||
| } | ||
|
|
||
| func (s *SnapshotSettings) Validate() error { | ||
| if s == nil { | ||
| return fmt.Errorf("settings is nil") | ||
| } | ||
|
|
||
| if s.Folder == "" { | ||
| return fmt.Errorf("folder is empty") | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package dynamo | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/aws/aws-sdk-go/service/dynamodb" | ||
| "github.com/aws/aws-sdk-go/service/dynamodbstreams" | ||
| "time" | ||
| ) | ||
|
|
||
| func NewMessageFromExport(item dynamodb.ItemResponse, keys []string, tableName string) (*Message, error) { | ||
| if len(item.Item) == 0 { | ||
| return nil, fmt.Errorf("item is nil or keys do not exist in this item payload") | ||
| } | ||
|
|
||
| if len(keys) == 0 { | ||
| return nil, fmt.Errorf("keys is nil") | ||
| } | ||
|
|
||
| // Snapshot time does not exist on the row | ||
| // Perhaps we can have it inferred from the manifest file in the future. | ||
| executionTime := time.Now() | ||
|
|
||
| rowData := transformNewImage(item.Item) | ||
| primaryKeys := make(map[string]interface{}) | ||
| for _, key := range keys { | ||
| val, isOk := rowData[key] | ||
| if !isOk { | ||
| return nil, fmt.Errorf("key does not exist in the item payload") | ||
| } | ||
|
|
||
| primaryKeys[key] = val | ||
| } | ||
|
|
||
| return &Message{ | ||
| op: "r", | ||
| tableName: tableName, | ||
| executionTime: executionTime, | ||
| rowData: rowData, | ||
| primaryKey: primaryKeys, | ||
| }, nil | ||
| } | ||
|
|
||
| func NewMessage(record *dynamodbstreams.Record, tableName string) (*Message, error) { | ||
| if record == nil || record.Dynamodb == nil { | ||
| return nil, fmt.Errorf("record is nil or dynamodb does not exist in this event payload") | ||
| } | ||
|
|
||
| if len(record.Dynamodb.Keys) == 0 { | ||
| return nil, fmt.Errorf("keys is nil") | ||
| } | ||
|
|
||
| executionTime := time.Now() | ||
| if record.Dynamodb.ApproximateCreationDateTime != nil { | ||
| executionTime = *record.Dynamodb.ApproximateCreationDateTime | ||
| } | ||
|
|
||
| op := "r" | ||
| if record.EventName != nil { | ||
| switch *record.EventName { | ||
| case "INSERT": | ||
| op = "c" | ||
| case "MODIFY": | ||
| op = "u" | ||
| case "REMOVE": | ||
| op = "d" | ||
| } | ||
| } | ||
|
|
||
| return &Message{ | ||
| op: op, | ||
| tableName: tableName, | ||
| executionTime: executionTime, | ||
| rowData: transformNewImage(record.Dynamodb.NewImage), | ||
| primaryKey: transformNewImage(record.Dynamodb.Keys), | ||
| }, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package dynamo | ||
|
|
||
| import ( | ||
| "github.com/artie-labs/transfer/lib/ptr" | ||
| "github.com/aws/aws-sdk-go/service/dynamodb" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func (d *DynamoDBTestSuite) Test_NewMessageFromExport() { | ||
| type _tc struct { | ||
| name string | ||
| item dynamodb.ItemResponse | ||
| keys []string | ||
| tableName string | ||
| expectedError string | ||
| } | ||
|
|
||
| tcs := []_tc{ | ||
| { | ||
| name: "Test with empty item", | ||
| item: dynamodb.ItemResponse{ | ||
| Item: map[string]*dynamodb.AttributeValue{}, | ||
| }, | ||
| keys: []string{"id"}, | ||
| tableName: "test", | ||
| expectedError: "item is nil or keys do not exist in this item payload", | ||
| }, | ||
| { | ||
| name: "Test with empty keys", | ||
| item: dynamodb.ItemResponse{ | ||
| Item: map[string]*dynamodb.AttributeValue{ | ||
| "id": { | ||
| S: ptr.ToString("1"), | ||
| }, | ||
| }, | ||
| }, | ||
| keys: []string{}, | ||
| tableName: "test", | ||
| expectedError: "keys is nil", | ||
| }, | ||
| { | ||
| name: "Test with valid item and keys", | ||
| item: dynamodb.ItemResponse{ | ||
| Item: map[string]*dynamodb.AttributeValue{ | ||
| "id": { | ||
| S: ptr.ToString("1"), | ||
| }, | ||
| }, | ||
| }, | ||
| keys: []string{"id"}, | ||
| tableName: "test", | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range tcs { | ||
| msg, err := NewMessageFromExport(tc.item, tc.keys, tc.tableName) | ||
| if tc.expectedError != "" { | ||
| assert.Equal(d.T(), tc.expectedError, err.Error(), tc.name) | ||
| } else { | ||
| assert.NoError(d.T(), err, tc.name) | ||
| assert.Equal(d.T(), tc.tableName, msg.tableName, tc.name) | ||
| assert.Equal(d.T(), "r", msg.op, tc.name) | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.