Skip to content

DynamoDBBuilder

Bases: AbstractAWSResourceBuilder['DynamoDBBuilder', DynamoDBConfig]

Builder for creating AWS DynamoDB tables with organizational standards.

Creates DynamoDB tables with configurable key schemas, Global Secondary Indexes, streaming capabilities, and automatic Parameter Store integration for cross-stack references. Supports pay-per-request billing and comprehensive tagging.

Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class DynamoDBBuilder(AbstractAWSResourceBuilder["DynamoDBBuilder", DynamoDBConfig]):
    """Builder for creating AWS DynamoDB tables with organizational standards.

    Creates DynamoDB tables with configurable key schemas, Global Secondary Indexes,
    streaming capabilities, and automatic Parameter Store integration for cross-stack
    references. Supports pay-per-request billing and comprehensive tagging.
    """

    _resource_type : AWSResourceType = AWSResourceType.DYNAMODB

    def reset(self) -> None:
        """Reset the builder to its initial state.

        Clears all internal state including table instance and table base name.
        Called internally to prepare for a new build.
        """
        super().reset()
        self._table_base_name: str = None
        self._table: dynamodb.Table = None

    def set_table_base_name(self, name: str) -> 'DynamoDBBuilder':
        """Set the base name for the DynamoDB table.

        The base name will be used to construct the final table name
        following organizational naming conventions.

        Args:
            name: Base name for the table

        Returns:
            Self for method chaining
        """
        self._table_base_name = name
        return self

    def build(self, scope: Construct) -> dynamodb.Table:
        """Build and return the configured DynamoDB table.

        Creates a DynamoDB table with partition key, optional sort key,
        Global Secondary Indexes, streaming configuration, and automatic
        Parameter Store entries for cross-stack references.

        Args:
            scope: CDK construct scope where the table will be created

        Returns:
            Configured DynamoDB Table instance

        Raises:
            ValidationError: If table configuration validation fails
            ValueError: If required table base name is not set
        """
        super().build()
        partition_key = dynamodb.Attribute(
            name=self._config.pk.name,
            type=self._config.pk.type.as_cdk
        )

        sort_key: dynamodb.Attribute = None
        if self._config.sk is not None:
            sort_key = dynamodb.Attribute(
                name=self._config.sk.name,
                type=self._config.sk.type.as_cdk
            ) 

        table_name = self._get_name_for_resource(self._table_base_name, max_length=AWSResourceNameLength.DYN_TABLE.value)
        self._table = dynamodb.Table(
            scope, self._get_cfn_logical_id(self._table_base_name),
            table_name=table_name,
            partition_key=partition_key,
            sort_key=sort_key,
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
            removal_policy=RemovalPolicy.RETAIN,
            stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES if self._config.enable_stream else None
        )

        if self._config.indexes is not None:
            for idx_name, idx_config in self._config.indexes.items():
                gsi_name = f"{idx_name.replace("_", "-")}-index"

                index_sort_key: dynamodb.Attribute = None
                if idx_config.sk is not None:
                    index_sort_key = dynamodb.Attribute(
                        name=idx_config.sk.name,
                        type=idx_config.sk.type.as_cdk
                    )

                self._table.add_global_secondary_index(
                    index_name=gsi_name,
                    partition_key=dynamodb.Attribute(
                        name=idx_config.pk.name,
                        type=idx_config.pk.type.as_cdk
                    ),
                    sort_key=index_sort_key,
                    projection_type=dynamodb.ProjectionType.ALL
                )

        # Store the table arn in the store parameter in order to be available to other stacks
        self._application_helper.store_parameter(scope, 
                                                f"DYNAMODB_TABLE_{table_name.replace("-", "_")}", 
                                                self._table.table_arn)
        if self._config.enable_stream:
            # Store the stream arn in the store parameter in order to be available to other stacks
            self._application_helper.store_parameter(scope, 
                                                f"DYNAMODB_TABLE_STREAM_{table_name.replace("-", "_")}", 
                                                self._table.table_stream_arn)

        # Add the tag to the table instance (It'll handle the backup strategy)
        Tags.of(self._table).add(self._config.tag_key, self._config.tag_value)

        self._tag_resource(self._table)

        return self._table

    def _set_config(self) -> None:
        """Create and validate the DynamoDB table configuration.

        Merges the base builder configuration with DynamoDB-specific parameters
        to create a validated DynamoDBConfig object. The configuration includes
        table naming, key schema, index definitions, streaming settings, and
        other DynamoDB-specific parameters required for table creation.

        The method combines:
            - Base configuration from self._builder_config (tags, environment, etc.)
            - DynamoDB-specific parameter: table_base_name from self._table_base_name

        Raises:
            ValidationError: If the DynamoDBConfig validation fails.
            ValueError: If table_base_name has not been set via set_table_base_name()

        Side Effects:
            Sets self._config to a validated DynamoDBConfig instance that can be
            accessed through the config property after this method completes.
        """
        try:
            self._config = DynamoDBConfig(**{
                **self._builder_config,
                "table_base_name": self._table_base_name
            })
        except ValidationError as e:
            self._log_validation_error(e, DynamoDBConfig)
            raise

    def _control_consistency(self) -> None:
        """Validate builder configuration and internal state consistency.

        Ensures table base name is set and validates the complete
        configuration using the DynamoDBConfig model.

        Raises:
            ValueError: If table base name is not set
            ValidationError: If configuration validation fails
        """
        super()._control_consistency()

        if not self._table_base_name:
            raise ValueError("Table base name must be set before building.")

        self._set_config()

Attributes

_resource_type = AWSResourceType.DYNAMODB class-attribute instance-attribute

Functions

_control_consistency()

Validate builder configuration and internal state consistency.

Ensures table base name is set and validates the complete configuration using the DynamoDBConfig model.

Raises:

Type Description
ValueError

If table base name is not set
ValidationError

If configuration validation fails
Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def _control_consistency(self) -> None:
    """Validate builder configuration and internal state consistency.

    Ensures table base name is set and validates the complete
    configuration using the DynamoDBConfig model.

    Raises:
        ValueError: If table base name is not set
        ValidationError: If configuration validation fails
    """
    super()._control_consistency()

    if not self._table_base_name:
        raise ValueError("Table base name must be set before building.")

    self._set_config()

_set_config()

Create and validate the DynamoDB table configuration.

Merges the base builder configuration with DynamoDB-specific parameters to create a validated DynamoDBConfig object. The configuration includes table naming, key schema, index definitions, streaming settings, and other DynamoDB-specific parameters required for table creation.

The method combines
  • Base configuration from self._builder_config (tags, environment, etc.)
  • DynamoDB-specific parameter: table_base_name from self._table_base_name

Raises:

Type Description
ValidationError

If the DynamoDBConfig validation fails.
ValueError

If table_base_name has not been set via set_table_base_name()
Side Effects

Sets self._config to a validated DynamoDBConfig instance that can be accessed through the config property after this method completes.

Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def _set_config(self) -> None:
    """Create and validate the DynamoDB table configuration.

    Merges the base builder configuration with DynamoDB-specific parameters
    to create a validated DynamoDBConfig object. The configuration includes
    table naming, key schema, index definitions, streaming settings, and
    other DynamoDB-specific parameters required for table creation.

    The method combines:
        - Base configuration from self._builder_config (tags, environment, etc.)
        - DynamoDB-specific parameter: table_base_name from self._table_base_name

    Raises:
        ValidationError: If the DynamoDBConfig validation fails.
        ValueError: If table_base_name has not been set via set_table_base_name()

    Side Effects:
        Sets self._config to a validated DynamoDBConfig instance that can be
        accessed through the config property after this method completes.
    """
    try:
        self._config = DynamoDBConfig(**{
            **self._builder_config,
            "table_base_name": self._table_base_name
        })
    except ValidationError as e:
        self._log_validation_error(e, DynamoDBConfig)
        raise

build(scope)

Build and return the configured DynamoDB table.

Creates a DynamoDB table with partition key, optional sort key, Global Secondary Indexes, streaming configuration, and automatic Parameter Store entries for cross-stack references.

Parameters:

Name Type Description Default
scope Construct

CDK construct scope where the table will be created

 required

Returns:

Type Description
Table

Configured DynamoDB Table instance

Raises:

Type Description
ValidationError

If table configuration validation fails
ValueError

If required table base name is not set
Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def build(self, scope: Construct) -> dynamodb.Table:
    """Build and return the configured DynamoDB table.

    Creates a DynamoDB table with partition key, optional sort key,
    Global Secondary Indexes, streaming configuration, and automatic
    Parameter Store entries for cross-stack references.

    Args:
        scope: CDK construct scope where the table will be created

    Returns:
        Configured DynamoDB Table instance

    Raises:
        ValidationError: If table configuration validation fails
        ValueError: If required table base name is not set
    """
    super().build()
    partition_key = dynamodb.Attribute(
        name=self._config.pk.name,
        type=self._config.pk.type.as_cdk
    )

    sort_key: dynamodb.Attribute = None
    if self._config.sk is not None:
        sort_key = dynamodb.Attribute(
            name=self._config.sk.name,
            type=self._config.sk.type.as_cdk
        ) 

    table_name = self._get_name_for_resource(self._table_base_name, max_length=AWSResourceNameLength.DYN_TABLE.value)
    self._table = dynamodb.Table(
        scope, self._get_cfn_logical_id(self._table_base_name),
        table_name=table_name,
        partition_key=partition_key,
        sort_key=sort_key,
        billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
        removal_policy=RemovalPolicy.RETAIN,
        stream=dynamodb.StreamViewType.NEW_AND_OLD_IMAGES if self._config.enable_stream else None
    )

    if self._config.indexes is not None:
        for idx_name, idx_config in self._config.indexes.items():
            gsi_name = f"{idx_name.replace("_", "-")}-index"

            index_sort_key: dynamodb.Attribute = None
            if idx_config.sk is not None:
                index_sort_key = dynamodb.Attribute(
                    name=idx_config.sk.name,
                    type=idx_config.sk.type.as_cdk
                )

            self._table.add_global_secondary_index(
                index_name=gsi_name,
                partition_key=dynamodb.Attribute(
                    name=idx_config.pk.name,
                    type=idx_config.pk.type.as_cdk
                ),
                sort_key=index_sort_key,
                projection_type=dynamodb.ProjectionType.ALL
            )

    # Store the table arn in the store parameter in order to be available to other stacks
    self._application_helper.store_parameter(scope, 
                                            f"DYNAMODB_TABLE_{table_name.replace("-", "_")}", 
                                            self._table.table_arn)
    if self._config.enable_stream:
        # Store the stream arn in the store parameter in order to be available to other stacks
        self._application_helper.store_parameter(scope, 
                                            f"DYNAMODB_TABLE_STREAM_{table_name.replace("-", "_")}", 
                                            self._table.table_stream_arn)

    # Add the tag to the table instance (It'll handle the backup strategy)
    Tags.of(self._table).add(self._config.tag_key, self._config.tag_value)

    self._tag_resource(self._table)

    return self._table

reset()

Reset the builder to its initial state.

Clears all internal state including table instance and table base name. Called internally to prepare for a new build.

Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
23
24
25
26
27
28
29
30
31
def reset(self) -> None:
    """Reset the builder to its initial state.

    Clears all internal state including table instance and table base name.
    Called internally to prepare for a new build.
    """
    super().reset()
    self._table_base_name: str = None
    self._table: dynamodb.Table = None

set_table_base_name(name)

Set the base name for the DynamoDB table.

The base name will be used to construct the final table name following organizational naming conventions.

Parameters:

Name Type Description Default
name str

Base name for the table

 required

Returns:

Type Description
DynamoDBBuilder

Self for method chaining
Source code in mare_aws_common_lib/builders/dynamodb_builder.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def set_table_base_name(self, name: str) -> 'DynamoDBBuilder':
    """Set the base name for the DynamoDB table.

    The base name will be used to construct the final table name
    following organizational naming conventions.

    Args:
        name: Base name for the table

    Returns:
        Self for method chaining
    """
    self._table_base_name = name
    return self