optional table-arn for ddb datasource

samtranslator/model/sam_resources.py

@@ -81,6 +81,7 @@
 from samtranslator.model.iam import IAMManagedPolicy, IAMRole, IAMRolePolicies
 from samtranslator.model.intrinsics import (
     fnGetAtt,
+    fnSub,
     is_intrinsic,
     is_intrinsic_if,
     is_intrinsic_no_value,
@@ -2604,7 +2605,7 @@ def _construct_ddb_datasources(
             cfn_datasource.DynamoDBConfig = self._parse_ddb_config(ddb_datasource)
 
             cfn_datasource.ServiceRoleArn, permissions_resources = self._parse_ddb_datasource_role(
-                ddb_datasource, cfn_datasource.get_runtime_attr("arn"), relative_id, datasource_logical_id, kwargs
+                ddb_datasource, cfn_datasource.get_runtime_attr("arn"), datasource_logical_id, kwargs
             )
 
             self._datasource_name_map[relative_id] = cfn_datasource.get_runtime_attr("name")
@@ -2617,21 +2618,16 @@ def _parse_ddb_datasource_role(
         self,
         ddb_datasource: aws_serverless_graphqlapi.DynamoDBDataSource,
         datasource_arn: Intrinsicable[str],
-        relative_id: str,
         datasource_logical_id: str,
         kwargs: Dict[str, Any],
     ) -> Tuple[str, List[Resource]]:
         # If the user defined a role, then there's no need to generate role/policy for them, so we return fast.
         if ddb_datasource.ServiceRoleArn:
             return cast(PassThrough, ddb_datasource.ServiceRoleArn), []
 
-        # If the user doesn't have their own role, then we will create for them if TableArn is defined.
-        table_arn = passthrough_value(
-            sam_expect(
-                ddb_datasource.TableArn, relative_id, f"DataSources.DynamoDb.{relative_id}.TableArn"
-            ).to_not_be_none(
-                "'TableArn' must be defined to create the role and policy if 'ServiceRoleArn' is not defined."
-            )
+        table_arn = cast(
+            Intrinsicable[str],
+            (ddb_datasource.TableArn if ddb_datasource.TableArn else self._compose_dynamodb_table_arn(ddb_datasource)),
         )
 
         permissions = ddb_datasource.Permissions or ["Read", "Write"]
@@ -2675,7 +2671,7 @@ def _parse_ddb_config(self, ddb_datasource: aws_serverless_graphqlapi.DynamoDBDa
     def _construct_ddb_datasource_connector_resources(
         datasource_id: str,
         source_arn: Intrinsicable[str],
-        destination_arn: str,
+        destination_arn: Intrinsicable[str],
         permissions: PermissionsType,
         role_name: Intrinsicable[str],
         kwargs: Dict[str, Any],
@@ -3026,3 +3022,15 @@ def _parse_appsync_resolver_functions(
     @staticmethod
     def _create_appsync_data_source_logical_id(api_id: str, data_source_type: str, data_source_relative_id: str) -> str:
         return f"{api_id}{data_source_relative_id}{data_source_type}DataSource"
+
+    @staticmethod
+    def _compose_dynamodb_table_arn(ddb_datasource: aws_serverless_graphqlapi.DynamoDBDataSource) -> Intrinsicable[str]:
+        return fnSub(
+            ArnGenerator.generate_dynamodb_table_arn(
+                partition="${AWS::Partition}", table_name="${__TableName__}", region="${__Region__}"
+            ),
+            {
+                "__TableName__": ddb_datasource.TableName,
+                "__Region__": passthrough_value(ddb_datasource.Region) or ref("AWS::Region"),
+            },
+        )

samtranslator/translator/arn_generator.py

@@ -34,19 +34,54 @@ class ArnGenerator:
 
     @classmethod
     def generate_arn(
-        cls, partition: str, service: str, resource: str, include_account_id: Optional[bool] = True
+        cls,
+        partition: str,
+        service: str,
+        resource: str,
+        include_account_id: bool = True,
+        region: Optional[str] = None,
     ) -> str:
+        """Generate AWS ARN.
+
+        Parameters
+        ----------
+        partition
+            AWS partition, ie "aws" or "aws-cn"
+        service
+            AWS service name
+        resource
+            Resource name, it must include service specific prefixes is "table/" for DynamoDB table
+        include_account_id, optional
+            include account ID in the ARN or not, by default True
+        region, optional
+            resource region, by default None.
+            To omit region in ARN (ie for a S3 bucket) pass "" (empty string).
+            Don't set it to any other default value because None can be passed by a caller and
+            must be handled in the function itself.
+
+        Returns
+        -------
+            Generated ARN
+
+        Raises
+        ------
+        RuntimeError
+            if service or resource are not provided
+        """
         if not service or not resource:
             raise RuntimeError("Could not construct ARN for resource.")
 
-        arn = "arn:{0}:{1}:${{AWS::Region}}:"
+        if region is None:
+            region = "${AWS::Region}"
+
+        arn = "arn:{0}:{1}:{region}:"
 
         if include_account_id:
             arn += "${{AWS::AccountId}}:"
 
         arn += "{2}"
 
-        return arn.format(partition, service, resource)
+        return arn.format(partition, service, resource, region=region)
 
     @classmethod
     def generate_aws_managed_policy_arn(cls, policy_name: str) -> str:
@@ -89,3 +124,24 @@ def get_partition_name(cls, region: Optional[str] = None) -> str:
             raise NoRegionFound("AWS Region cannot be found")
 
         return _region_to_partition(region)
+
+    @classmethod
+    def generate_dynamodb_table_arn(cls, partition: str, region: str, table_name: str) -> str:
+        """Generate DynamoDB table ARN.
+
+        Parameters
+        ----------
+        partition
+            _description_
+        region
+            DynamoDB table region
+        table_name
+            DynamoDB table name
+
+        Returns
+        -------
+            DynamoDB table ARN.
+        """
+        return ArnGenerator.generate_arn(
+            partition=partition, service="dynamodb", resource=f"table/{table_name}", region=region
+        )

tests/translator/input/error_graphqlapi.yaml

@@ -65,25 +65,6 @@ Resources:
         ExcludeVerboseContent: true
         NotSupposedToBeHere: yo
 
-  DataSourceNoServiceRoleArnOrTableArn:
-    Type: AWS::Serverless::GraphQLApi
-    Properties:
-      Name: SomeApi
-      SchemaInline: |
-        type Mutation {
-          addTodo(id: ID!, name: String, description: String, priority: Int): Todo
-        }
-      XrayEnabled: true
-      Auth:
-        Type: AWS_IAM
-      Tags:
-        key1: value1
-        key2: value2
-      DataSources:
-        DynamoDb:
-          MyDataSource:
-            TableName: some-table
-
   IdDefinedWithOtherProperties:
     Type: AWS::Serverless::GraphQLApi
     Properties:

tests/translator/input/graphqlapi_ddb_datasource_no_table_arn.yaml

@@ -0,0 +1,45 @@
+Transform: AWS::Serverless-2016-10-31
+
+Mappings:
+  TablesToRegions:
+    Table1:
+      Region: us-west-2
+
+Resources:
+  Table1:
+    Type: AWS::Serverless::SimpleTable
+    Properties:
+      TableName: Table1
+
+  SuperCoolAPI:
+    Type: AWS::Serverless::GraphQLApi
+    Properties:
+      SchemaInline: |
+        type Todo {
+          id: ID!
+          description: String!
+        }
+        type Mutation {
+          addTodo(id: ID!, description: String!): Todo!
+        }
+        type Query {
+          getTodo(id: ID!): Todo
+        }
+        schema {
+          mutation: Mutation
+          query: Query
+        }
+      Auth:
+        Type: AWS_IAM
+      DataSources:
+        DynamoDb:
+          HardcodedTableName:
+            TableName: Table1
+          TableNameIntrinsic:
+            TableName: !Ref Table1
+          RegionHardcoded:
+            TableName: Table1
+            Region: us-west-2
+          RegionIntrinsic:
+            TableName: !Ref Table1
+            Region: !FindInMap [TablesToRegions, Table1, Region]