migrator

README

Migrator

The Migrator is a tool to execute schema migrations of dynamodb tables.

Requirements

The Migrator tool requires on a dedicated Metadata dynamodb table. The migration tool should have the following permissions on the migration metadata table :

• dynamodb:GetItem
• dynamodb:PutItem
• dynamodb:UpdateItem

Examples

Execute a schema migration

const migrationMetadataTable = "migrationMetadata"

func ExecuteMigrations(ctx context.Context, client *dynamodb.Client) error {
	migrator := migrator.NewMigrator(migrationMetadataTable, 0, migrator.Migration{
		Name: "InitialMigration",
		Description: "Initial migration",
		MigratorFn: func(_ context.Context, _ migrator.DynamoDBClient) error {
			return nil
        }
    },
	    migrator.Must(migrator.NewScanAndUpdateMigration("SecondMigration", "Second migration", "tableToMigrate", 
			func(ctx context.Context, item map[string]types.AttributeValue) *dynamodb.UpdateItemInput) {
                //implement the migration for an attribute here
            }
        ),
    }
	
	return migrator.Execute(ctx, client)

Documentation

Index

• type DynamodbClient
• type Migration
  • func Must(migration Migration, err error) Migration
  • func NewScanAndUpdateMigration(name string, description string, table string, ...) (*Migration, error)
• type Migrator
  • func NewMigrator(migrationTableName string, migrationOffsetId uint64, migrations ...Migration) *Migrator
  • func (m *Migrator) Execute(ctx context.Context, client DynamodbClient) error
• type OptionFn
  • func ScanAndUpdateMigrationWithConsistentRead(consistentRead bool) OptionFn
  • func ScanAndUpdateMigrationWithFilterExpression(filter conditionexpression.ExpressionItem) OptionFn
  • func ScanAndUpdateMigrationWithMetadata(metadata map[string]interface{}) OptionFn
• type ScanAndUpdateMigrationOptions

Types

type DynamodbClient

type DynamodbClient interface {
	PutItem(ctx context.Context, params *dynamodb.PutItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.PutItemOutput, error)
	GetItem(ctx context.Context, params *dynamodb.GetItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.GetItemOutput, error)
	DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.DeleteItemOutput, error)
	Scan(ctx context.Context, params *dynamodb.ScanInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.ScanOutput, error)
	TransactWriteItems(ctx context.Context, params *dynamodb.TransactWriteItemsInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.TransactWriteItemsOutput, error)
	Query(ctx context.Context, params *dynamodb.QueryInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.QueryOutput, error)
	UpdateItem(ctx context.Context, params *dynamodb.UpdateItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.UpdateItemOutput, error)
}

type Migration

type Migration struct {
	// Name of the migration
	Name string

	// Description of the migration
	Description string

	// Migration function
	MigratorFn func(ctx context.Context, client DynamodbClient) error

	// JobMetadata to store in the migration metadata table
	JobMetadata map[string]interface{}
}

Migration to execute

func Must

func Must(migration Migration, err error) Migration

func NewScanAndUpdateMigration

func NewScanAndUpdateMigration(name string, description string, table string, updateFn func(ctx context.Context, item map[string]types.AttributeValue) *dynamodb.UpdateItemInput, optFn ...OptionFn) (*Migration, error)

NewScanAndUpdateMigration create a migration that will execute a scan. For each item in the table an updateFn will be executed. If the updateFn return a non nil *dynamodb.UpdateItemInput, the update will be executed.

type Migrator

type Migrator struct {
	MigrationIdOffset  uint64
	MigrationTableName string
	Migrations         []Migration
}

func NewMigrator

func NewMigrator(migrationTableName string, migrationOffsetId uint64, migrations ...Migration) *Migrator

NewMigrator creates a new Migrator that can execute a migration. The following parameters are used:

• migrationTableName: specifies the migration metadata table
• migrationOffsetId: requires to ID offset that should be used. IDs are automatically generated based on the order of the parameter 'migrations'. If you want to remove old migrations you need to specify the offset so new IDs are correctly generated.
• migrations: are the migrations to be executed

Migration with an ID that are already successful executed will be skipped. Note that migrations should be idempotent (towards itself). If a migration fails or the update of the metadata table fails, the migration will be retried in the next execution.

func (*Migrator) Execute

func (m *Migrator) Execute(ctx context.Context, client DynamodbClient) error

type OptionFn

type OptionFn func(*ScanAndUpdateMigrationOptions)

func ScanAndUpdateMigrationWithConsistentRead

func ScanAndUpdateMigrationWithConsistentRead(consistentRead bool) OptionFn

ScanAndUpdateMigrationWithConsistentRead annotate consistentRead on ScanAndUpdateMigration. Note that default consistent read will be true

func ScanAndUpdateMigrationWithFilterExpression

func ScanAndUpdateMigrationWithFilterExpression(filter conditionexpression.ExpressionItem) OptionFn

ScanAndUpdateMigrationWithFilterExpression set filter condition on ScanAndUpdateMigration

func ScanAndUpdateMigrationWithMetadata

func ScanAndUpdateMigrationWithMetadata(metadata map[string]interface{}) OptionFn

ScanAndUpdateMigrationWithMetadata add metadata to ScanAndUpdateMigration

type ScanAndUpdateMigrationOptions

type ScanAndUpdateMigrationOptions struct {
	FilterExpression conditionexpression.ExpressionItem
	ConsistentRead   *bool
	Metadata         map[string]interface{}
}