test updates; more docs; small changes to table options

Author: raduendv

feature/dynamodb/enhancedclient/batch_read.go

@@ -32,7 +32,7 @@ type BatchGetOperation[T any] struct {
 func (b *BatchGetOperation[T]) AddReadItem(item *T) error {
 	m, err := b.schema.createKeyMap(item)
 	if err != nil {
-		return fmt.Errorf(`error calling schema.createKeyMap: %v`, err)
+		return fmt.Errorf("error calling schema.createKeyMap: %w", err)
 	}
 
 	b.queue = append(b.queue, batchReadQueueItem{
@@ -62,6 +62,12 @@ func (b *BatchGetOperation[T]) AddReadItemByMap(m Map) error {
 //	seq := op.Execute(ctx)
 //	for res := range iter.Chan(seq) { ... }
 func (b *BatchGetOperation[T]) Execute(ctx context.Context, optFns ...func(options *dynamodb.Options)) iter.Seq[ItemResult[T]] {
+	var consecutiveErrors uint = 0
+	var maxConsecutiveErrors = b.table.options.MaxConsecutiveErrors
+	if maxConsecutiveErrors == 0 {
+		maxConsecutiveErrors = DefaultMaxConsecutiveErrors
+	}
+
 	return func(yield func(ItemResult[T]) bool) {
 		tableName := b.schema.TableName()
 		if tableName == nil {
@@ -87,8 +93,17 @@ func (b *BatchGetOperation[T]) Execute(ctx context.Context, optFns ...func(optio
 
 			res, err := b.client.BatchGetItem(ctx, bgii, optFns...)
 			if err != nil {
-				yield(ItemResult[T]{err: err})
+				if !yield(ItemResult[T]{err: err}) {
+					return
+				}
+
+				if consecutiveErrors >= maxConsecutiveErrors {
+					return
+				} else {
+					continue
+				}
 			}
+			consecutiveErrors = 0
 
 			if res != nil && res.Responses != nil {
 				for _, item := range res.Responses[*tableName] {

feature/dynamodb/enhancedclient/batch_write.go

@@ -36,12 +36,12 @@ type BatchWriteOperation[T any] struct {
 // The item is encoded using the table's schema and extensions are applied before writing.
 func (b *BatchWriteOperation[T]) AddPut(item *T) error {
 	if err := b.table.applyBeforeWriteExtensions(item); err != nil {
-		return fmt.Errorf(`error calling table.applyBeforeWriteExtensions(): %v`, err)
+		return fmt.Errorf("error calling table.applyBeforeWriteExtensions(): %w", err)
 	}
 
 	m, err := b.schema.Encode(item)
 	if err != nil {
-		return fmt.Errorf(`error calling schema.Encode(): %v`, err)
+		return fmt.Errorf("error calling schema.Encode(): %w", err)
 	}
 
 	b.queue = append(b.queue, batchWriteQueueItem{
@@ -72,7 +72,7 @@ func (b *BatchWriteOperation[T]) AddRawPut(i map[string]types.AttributeValue) er
 func (b *BatchWriteOperation[T]) AddDelete(item *T) error {
 	m, err := b.schema.createKeyMap(item)
 	if err != nil {
-		return fmt.Errorf(`error calling schema.createKeyMap(): %v`, err)
+		return fmt.Errorf("error calling schema.createKeyMap(): %w", err)
 	}
 
 	b.queue = append(b.queue, batchWriteQueueItem{
@@ -102,6 +102,12 @@ func (b *BatchWriteOperation[T]) AddRawDelete(i map[string]types.AttributeValue)
 // It sends requests in batches of up to 25 items, and retries unprocessed items until all are written.
 // Returns an error if the table name is not set or if a request fails.
 func (b *BatchWriteOperation[T]) Execute(ctx context.Context, optFns ...func(options *dynamodb.Options)) error {
+	var consecutiveErrors uint = 0
+	var maxConsecutiveErrors = b.table.options.MaxConsecutiveErrors
+	if maxConsecutiveErrors == 0 {
+		maxConsecutiveErrors = DefaultMaxConsecutiveErrors
+	}
+
 	tableName := b.schema.TableName()
 	if tableName == nil {
 		return fmt.Errorf("empty table name, did you forget Schema[T].WithName()?")
@@ -129,7 +135,7 @@ func (b *BatchWriteOperation[T]) Execute(ctx context.Context, optFns ...func(opt
 					},
 				})
 			default:
-				return fmt.Errorf("uknown batchWriteOpType: %d", op.typ)
+				return fmt.Errorf("unknown batchWriteOpType: %d", op.typ)
 			}
 		}
 
@@ -141,22 +147,31 @@ func (b *BatchWriteOperation[T]) Execute(ctx context.Context, optFns ...func(opt
 
 		res, err := b.client.BatchWriteItem(ctx, bwii, optFns...)
 		if err != nil {
-			return err
+			consecutiveErrors++
+
+			if consecutiveErrors >= maxConsecutiveErrors {
+				return err
+			}
 		}
 
+		var unprocessedItems []types.WriteRequest
 		if res != nil && res.UnprocessedItems != nil {
-			for _, ui := range res.UnprocessedItems[*tableName] {
-				if ui.PutRequest != nil {
-					b.queue = append(b.queue, batchWriteQueueItem{
-						typ:  batchWriteOpPut,
-						item: ui.PutRequest.Item,
-					})
-				} else if ui.DeleteRequest != nil {
-					b.queue = append(b.queue, batchWriteQueueItem{
-						typ:  batchWriteOpDelete,
-						item: ui.DeleteRequest.Key,
-					})
-				}
+			unprocessedItems = res.UnprocessedItems[*tableName]
+		} else if err != nil {
+			unprocessedItems = bwii.RequestItems[*tableName]
+		}
+
+		for _, ui := range unprocessedItems {
+			if ui.PutRequest != nil {
+				b.queue = append(b.queue, batchWriteQueueItem{
+					typ:  batchWriteOpPut,
+					item: ui.PutRequest.Item,
+				})
+			} else if ui.DeleteRequest != nil {
+				b.queue = append(b.queue, batchWriteQueueItem{
+					typ:  batchWriteOpDelete,
+					item: ui.DeleteRequest.Key,
+				})
 			}
 		}
 	}

feature/dynamodb/enhancedclient/schema_test.go

@@ -19,8 +19,8 @@ func TestSchema(t *testing.T) {
 		t.Fatalf("NewSchema error: %v", err)
 	}
 
-	if len(actual.cachedFields.fields) != 17 {
-		t.Fatalf("expected %d CachedFields, found %d", 17, len(actual.cachedFields.fields))
+	if len(actual.cachedFields.fields) != 18 {
+		t.Fatalf("expected %d CachedFields, found %d", 18, len(actual.cachedFields.fields))
 	}
 
 	// | Index Name          | Partition Key | Sort Key     | Type | Notes                       |
@@ -327,6 +327,15 @@ func TestSchema(t *testing.T) {
 						Name: "last_name",
 					},
 				},
+				{
+					Name:        "nick_name",
+					NameFromTag: true,
+					Index:       []int{16},
+					Type:        reflect.TypeFor[string](),
+					Tag: Tag{
+						Name: "nick_name",
+					},
+				},
 			},
 			fieldsByName: map[string]int{
 				"order_id":      0,
@@ -346,6 +355,7 @@ func TestSchema(t *testing.T) {
 				"note":          14,
 				"first_name":    15,
 				"last_name":     16,
+				"nick_name":     17,
 			},
 		},
 		enc: NewEncoder[order](),
@@ -474,6 +484,9 @@ func TestSchemaEncodeDecode(t *testing.T) {
 		"last_name": &types.AttributeValueMemberS{
 			Value: "1e73f306-3362-49da-af74-41e1befff588",
 		},
+		"nick_name": &types.AttributeValueMemberNULL{
+			Value: true,
+		},
 	}
 
 	s, _ := NewSchema[order]()

feature/dynamodb/enhancedclient/structs_test.go

@@ -22,6 +22,7 @@ type order struct {
 	customerNote      string   `dynamodbav:"note" dynamodbgetter:"Note" dynamodbsetter:"SetNote" dynamodbindex:"NoteIndex,global,partition"`
 	CustomerFirstName string   `dynamodbav:"first_name"`
 	CustomerLastName  string   `dynamodbav:"last_name"`
+	CustomerNickName  *string  `dynamodbav:"nick_name"`
 }
 
 // Getter method for customerNote

feature/dynamodb/enhancedclient/table.go

@@ -7,17 +7,44 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 )
 
+// TableOptions provides configuration options for a DynamoDB Table.
+//
+// T is the type of the items stored in the table.
 type TableOptions[T any] struct {
-	Schema            *Schema[T]
-	DynamoDBOptions   []func(*dynamodb.Options)
+	// Schema defines the schema for the table, including attribute mapping and validation.
+	Schema *Schema[T]
+
+	// DynamoDBOptions is a list of functions to customize the underlying DynamoDB client options.
+	DynamoDBOptions []func(*dynamodb.Options)
+
+	// ExtensionRegistry holds the registry of extensions to be used with the table.
 	ExtensionRegistry *ExtensionRegistry[T]
+
+	// MaxConsecutiveErrors sets the maximum number of consecutive errors allowed during batch, query, or scan operations.
+	// If this threshold is exceeded, the operation will stop and return.
+	// If set to 0, the default value of DefaultMaxConsecutiveErrors will be used.
+	MaxConsecutiveErrors uint
 }
 
+// DefaultMaxConsecutiveErrors is the fallback value used for MaxConsecutiveErrors when it is set to 0.
+// A value of 1 means the operation will stop after the first error.
+const DefaultMaxConsecutiveErrors uint = 1
+
+// Table represents a strongly-typed DynamoDB table for items of type T.
+//
+// It provides methods for interacting with DynamoDB using the provided client and options.
 type Table[T any] struct {
-	client  Client
+	// client is the DynamoDB client used to perform operations on the table.
+	client Client
+
+	// options holds the configuration options for the table.
 	options TableOptions[T]
 }
 
+// NewTable creates a new Table for items of type T using the provided client and configuration functions.
+//
+// The configuration functions can be used to customize the TableOptions before the table is created.
+// Returns an error if T is not a struct type or if required options cannot be resolved.
 func NewTable[T any](client Client, fns ...func(options *TableOptions[T])) (*Table[T], error) {
 	if reflect.TypeFor[T]().Kind() != reflect.Struct {
 		return nil, fmt.Errorf("NewClient() can only be created from structs, %T given", *new(T))
@@ -32,6 +59,7 @@ func NewTable[T any](client Client, fns ...func(options *TableOptions[T])) (*Tab
 	defaultResolvers := []resolverFn[T]{
 		resolveDefaultSchema[T],
 		resolveDefaultExtensionRegistry[T],
+		resolveDefaultMaxConsecutiveErrors[T],
 	}
 
 	for _, fn := range defaultResolvers {
@@ -46,20 +74,44 @@ func NewTable[T any](client Client, fns ...func(options *TableOptions[T])) (*Tab
 	}, nil
 }
 
+// WithSchema returns a configuration function that sets the Schema for TableOptions.
+//
+// Use this to specify a custom schema when creating a Table.
 func WithSchema[T any](schema *Schema[T]) func(options *TableOptions[T]) {
 	return func(options *TableOptions[T]) {
 		options.Schema = schema
 	}
 }
 
+// WithExtensionRegistry returns a configuration function that sets the ExtensionRegistry for TableOptions.
+//
+// Use this to specify a custom extension registry when creating a Table.
 func WithExtensionRegistry[T any](registry *ExtensionRegistry[T]) func(options *TableOptions[T]) {
 	return func(options *TableOptions[T]) {
 		options.ExtensionRegistry = registry
 	}
 }
 
+// WithMaxConsecutiveErrors returns a configuration function that sets the MaxConsecutiveErrors option for TableOptions.
+//
+// Use this to specify the maximum number of consecutive errors allowed during batch, query, or scan operations.
+// A value of 0 means no limit is enforced.
+// WithMaxConsecutiveErrors returns a configuration function that sets the MaxConsecutiveErrors option for TableOptions.
+//
+// Use this to specify the maximum number of consecutive errors allowed during batch, query, or scan operations.
+// If set to 0, the default value of DefaultMaxConsecutiveErrors will be used.
+func WithMaxConsecutiveErrors[T any](maxConsecutiveErrors uint) func(options *TableOptions[T]) {
+	return func(options *TableOptions[T]) {
+		options.MaxConsecutiveErrors = maxConsecutiveErrors
+	}
+}
+
+// resolverFn defines a function type for resolving or setting default options on TableOptions.
 type resolverFn[T any] func(opts *TableOptions[T]) error
 
+// resolveDefaultSchema sets a default schema on TableOptions if none is provided.
+//
+// Returns an error if the schema cannot be created.
 func resolveDefaultSchema[T any](opts *TableOptions[T]) error {
 	if opts.Schema == nil {
 		var err error
@@ -72,10 +124,21 @@ func resolveDefaultSchema[T any](opts *TableOptions[T]) error {
 	return nil
 }
 
+// resolveDefaultExtensionRegistry sets a default extension registry on TableOptions if none is provided.
 func resolveDefaultExtensionRegistry[T any](opts *TableOptions[T]) error {
 	if opts.ExtensionRegistry == nil {
 		opts.ExtensionRegistry = DefaultExtensionRegistry[T]()
 	}
 
 	return nil
 }
+
+// resolveDefaultMaxConsecutiveErrors sets MaxConsecutiveErrors to DefaultMaxConsecutiveErrors
+// if it is not explicitly set (i.e., if the value is 0).
+// This ensures a sensible default for error handling in batch, query, or scan operations.
+func resolveDefaultMaxConsecutiveErrors[T any](opts *TableOptions[T]) error {
+	if opts.MaxConsecutiveErrors == 0 {
+		opts.MaxConsecutiveErrors = DefaultMaxConsecutiveErrors
+	}
+	return nil
+}