Source code for ecs_composex.dynamodb.dynamodb_stack
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 John Mille <john@compose-x.io>
"""
Module to create the root stack for DynamoDB tables
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ecs_composex.common.settings import ComposeXSettings
from ecs_composex.mods_manager import XResourceModule
from botocore.exceptions import ClientError
from compose_x_common.aws.dynamodb import TABLE_ARN_RE
from compose_x_common.compose_x_common import attributes_to_mapping, keyisset
from troposphere import GetAtt, Ref
from troposphere.dynamodb import Table as CfnTable
from ecs_composex.common.logging import LOG
from ecs_composex.common.stacks import ComposeXStack
from ecs_composex.common.troposphere_tools import build_template
from ecs_composex.compose.x_resources.api_x_resources import ApiXResource
from ecs_composex.dynamodb.dynamodb_params import TABLE_ARN, TABLE_NAME
from ecs_composex.dynamodb.dynamodb_template import create_dynamodb_template
[docs]def get_dynamodb_table_config(table, account_id, resource_id):
"""
:param Table table:
:param str account_id:
:param str resource_id:
:return:
"""
table_attributes_mapping = {
TABLE_NAME: "TableName",
TABLE_ARN: "TableArn",
}
client = table.lookup_session.client("dynamodb")
try:
table_r = client.describe_table(TableName=resource_id)["Table"]
table_config = attributes_to_mapping(table_r, table_attributes_mapping)
return table_config
# except client.exceptions.ResourceNotFoundException:
# raise
except ClientError as error:
LOG.error(error)
raise
[docs]class Table(ApiXResource):
"""
Class to represent a DynamoDB Table
"""
def __init__(
self,
name: str,
definition: dict,
module: XResourceModule,
settings: ComposeXSettings,
):
super().__init__(name, definition, module, settings)
self.arn_parameter = TABLE_ARN
self.ref_parameter = TABLE_NAME
self.default_cloudmap_settings = {
"ReturnValues": {
TABLE_NAME.title: TABLE_NAME.title,
TABLE_ARN.title: TABLE_ARN.title,
}
}
[docs] def init_outputs(self):
self.output_properties = {
TABLE_NAME: (self.logical_name, self.cfn_resource, Ref, None),
TABLE_ARN: (
f"{self.logical_name}{TABLE_ARN.title}",
self.cfn_resource,
GetAtt,
TABLE_ARN.return_value,
),
}
[docs]def resolve_lookup(
lookup_resources: list[Table], settings: ComposeXSettings, module: XResourceModule
) -> None:
"""
Lookup AWS Resource
:param list[Table] lookup_resources:
:param ecs_composex.common.settings.ComposeXSettings settings:
:param XResourceModule module:
"""
if not keyisset(module.mapping_key, settings.mappings):
settings.mappings[module.mapping_key] = {}
for resource in lookup_resources:
resource.lookup_resource(
TABLE_ARN_RE,
get_dynamodb_table_config,
CfnTable.resource_type,
"dynamodb:table",
)
LOG.info(f"{module.res_key}.{resource.name} - Matched to {resource.arn}")
settings.mappings[module.mapping_key].update(
{resource.logical_name: resource.mappings}
)
[docs]class XStack(ComposeXStack):
"""
Class for Dynamodb
"""
def __init__(
self, title, settings: ComposeXSettings, module: XResourceModule, **kwargs
):
if module.lookup_resources:
resolve_lookup(module.lookup_resources, settings, module)
if module.new_resources:
stack_template = build_template("Root template for DynamoDB tables")
super().__init__(title, stack_template, **kwargs)
create_dynamodb_template(module.new_resources, stack_template, self)
else:
self.is_void = True
for resource in module.resources_list:
if resource.lookup:
resource.stack = self