diff --git a/samtranslator/model/__init__.py b/samtranslator/model/__init__.py index 400fae78f..e53f8f716 100644 --- a/samtranslator/model/__init__.py +++ b/samtranslator/model/__init__.py @@ -193,7 +193,7 @@ def from_dict(cls, logical_id, resource_dict, relative_id=None, sam_plugins=None if attr in resource_dict: resource.set_resource_attribute(attr, resource_dict[attr]) - resource.validate_properties() # type: ignore[no-untyped-call] + resource.validate_properties() return resource @classmethod @@ -231,7 +231,7 @@ def _validate_resource_dict(cls, logical_id, resource_dict): # type: ignore[no- if "Properties" in resource_dict and not isinstance(resource_dict["Properties"], dict): raise InvalidResourceException(logical_id, "Properties of a resource must be an object.") - def to_dict(self): # type: ignore[no-untyped-def] + def to_dict(self) -> Dict[str, Dict[str, Any]]: """Validates that the required properties for this Resource have been provided, then returns a dict corresponding to the given Resource object. This dict will take the format of a single entry in the Resources section of a CloudFormation template, and will take the following format. :: @@ -252,7 +252,7 @@ def to_dict(self): # type: ignore[no-untyped-def] :rtype: dict :raises TypeError: if a required property is missing from this Resource """ - self.validate_properties() # type: ignore[no-untyped-call] + self.validate_properties() resource_dict = self._generate_resource_dict() # type: ignore[no-untyped-call] @@ -300,7 +300,7 @@ def __setattr__(self, name, value): # type: ignore[no-untyped-def] ), ) - def validate_properties(self): # type: ignore[no-untyped-def] + def validate_properties(self) -> None: """Validates that the required properties for this Resource have been populated, and that all properties have valid values. @@ -523,7 +523,7 @@ class ResourceTypeResolver(object): """ResourceTypeResolver maps Resource Types to Resource classes, e.g. AWS::Serverless::Function to samtranslator.model.sam_resources.SamFunction.""" - def __init__(self, *modules): # type: ignore[no-untyped-def] + def __init__(self, *modules: Any): """Initializes the ResourceTypeResolver from the given modules. :param modules: one or more Python modules containing Resource definitions diff --git a/samtranslator/model/lambda_.py b/samtranslator/model/lambda_.py index 71d5ec542..0ffa41a14 100644 --- a/samtranslator/model/lambda_.py +++ b/samtranslator/model/lambda_.py @@ -1,4 +1,4 @@ -from typing import Optional, Dict, Any, List +from typing import Optional, Dict, Any, List, Union from samtranslator.model import PropertyType, Resource from samtranslator.model.types import IS_DICT, is_type, one_of, IS_STR, list_of, any_type from samtranslator.model.intrinsics import fnGetAtt, ref @@ -152,6 +152,13 @@ class LambdaLayerVersion(Resource): "LicenseInfo": PropertyType(False, IS_STR), } + Content: Dict[str, Any] + Description: Optional[Intrinsicable[str]] + LayerName: Optional[Intrinsicable[str]] + CompatibleArchitectures: Optional[List[Union[str, Dict[str, Any]]]] + CompatibleRuntimes: Optional[List[Union[str, Dict[str, Any]]]] + LicenseInfo: Optional[Intrinsicable[str]] + runtime_attrs = {"name": lambda self: ref(self.logical_id), "arn": lambda self: fnGetAtt(self.logical_id, "Arn")} diff --git a/samtranslator/model/s3_utils/uri_parser.py b/samtranslator/model/s3_utils/uri_parser.py index df91b0137..c57a556ec 100644 --- a/samtranslator/model/s3_utils/uri_parser.py +++ b/samtranslator/model/s3_utils/uri_parser.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Union from urllib.parse import urlparse, parse_qs from samtranslator.model.exceptions import InvalidResourceException @@ -62,7 +62,9 @@ def construct_image_code_object(image_uri, logical_id, property_name): # type: return {"ImageUri": image_uri} -def construct_s3_location_object(location_uri, logical_id, property_name): # type: ignore[no-untyped-def] +def construct_s3_location_object( + location_uri: Union[str, Dict[str, Any]], logical_id: str, property_name: str +) -> Dict[str, Any]: """Constructs a Lambda `Code` or `Content` property, from the SAM `CodeUri` or `ContentUri` property. This follows the current scheme for Lambda Functions and LayerVersions. diff --git a/samtranslator/model/sam_resources.py b/samtranslator/model/sam_resources.py index 25018ab8f..2085f1da4 100644 --- a/samtranslator/model/sam_resources.py +++ b/samtranslator/model/sam_resources.py @@ -1,6 +1,6 @@ """ SAM macro definitions """ import copy -from typing import Any, cast, Dict, List, Optional, Tuple, Union +from typing import Any, cast, Dict, List, Optional, Tuple, Union, Callable from samtranslator.intrinsics.resolver import IntrinsicsResolver from samtranslator.feature_toggle.feature_toggle import FeatureToggle from samtranslator.model.connector.connector import ( @@ -167,7 +167,7 @@ class SamFunction(SamResourceMacro): SnapStart: Optional[Dict[str, Any]] FunctionUrlConfig: Optional[Dict[str, Any]] - event_resolver = ResourceTypeResolver( # type: ignore[no-untyped-call] + event_resolver = ResourceTypeResolver( samtranslator.model.eventsources, samtranslator.model.eventsources.pull, samtranslator.model.eventsources.push, @@ -191,9 +191,9 @@ class SamFunction(SamResourceMacro): "DestinationQueue": SQSQueue.resource_type, } - def resources_to_link(self, resources): # type: ignore[no-untyped-def] + def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]: try: - return {"event_resources": self._event_resources_to_link(resources)} # type: ignore[no-untyped-call] + return {"event_resources": self._event_resources_to_link(resources)} except InvalidEventException as e: raise InvalidResourceException(self.logical_id, e.message) @@ -260,7 +260,7 @@ def to_cloudformation(self, **kwargs): # type: ignore[no-untyped-def] feature_toggle, ) - event_invoke_policies = [] + event_invoke_policies: List[Dict[str, Any]] = [] if self.EventInvokeConfig: function_name = lambda_function.logical_id event_invoke_resources, event_invoke_policies = self._construct_event_invoke_config( @@ -279,7 +279,7 @@ def to_cloudformation(self, **kwargs): # type: ignore[no-untyped-def] resources.append(execution_role) try: - resources += self._generate_event_resources( # type: ignore[no-untyped-call] + resources += self._generate_event_resources( lambda_function, execution_role, kwargs["event_resources"], @@ -291,7 +291,15 @@ def to_cloudformation(self, **kwargs): # type: ignore[no-untyped-def] return resources - def _construct_event_invoke_config(self, function_name, alias_name, lambda_alias, intrinsics_resolver, conditions, event_invoke_config: Dict[str, Any]): # type: ignore[no-untyped-def] + def _construct_event_invoke_config( + self, + function_name: str, + alias_name: str, + lambda_alias: Optional[LambdaAlias], + intrinsics_resolver: IntrinsicsResolver, + conditions: Any, + event_invoke_config: Dict[str, Any], + ) -> Tuple[List[Any], List[Dict[str, Any]]]: """ Create a `AWS::Lambda::EventInvokeConfig` based on the input dict `EventInvokeConfig` """ @@ -502,7 +510,7 @@ def _construct_lambda_function(self) -> LambdaFunction: lambda_function.VpcConfig = self.VpcConfig lambda_function.Role = self.Role lambda_function.Environment = self.Environment - lambda_function.Code = self._construct_code_dict() # type: ignore[no-untyped-call] + lambda_function.Code = self._construct_code_dict() lambda_function.KmsKeyArn = self.KmsKeyArn lambda_function.ReservedConcurrentExecutions = self.ReservedConcurrentExecutions lambda_function.Tags = self._construct_tag_list(self.Tags) @@ -694,7 +702,7 @@ def _validate_dlq(self, dead_letter_queue: Dict[str, Any]) -> None: self.logical_id, "'DeadLetterQueue' requires Type of {}".format(valid_dlq_types) ) - def _event_resources_to_link(self, resources): # type: ignore[no-untyped-def] + def _event_resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]: event_resources = {} if self.Events: for logical_id, event_dict in self.Events.items(): @@ -708,7 +716,7 @@ def _event_resources_to_link(self, resources): # type: ignore[no-untyped-def] return event_resources @staticmethod - def order_events(event): # type: ignore[no-untyped-def] + def order_events(event: Tuple[str, Any]) -> Any: """ Helper method for sorting Function Events. Returns a key to use in sorting this event @@ -722,9 +730,14 @@ def order_events(event): # type: ignore[no-untyped-def] return logical_id return event_dict.get("Properties", {}).get("Path", logical_id) - def _generate_event_resources( # type: ignore[no-untyped-def] - self, lambda_function, execution_role, event_resources, intrinsics_resolver, lambda_alias=None - ): + def _generate_event_resources( + self, + lambda_function: LambdaFunction, + execution_role: Optional[IAMRole], + event_resources: Any, + intrinsics_resolver: IntrinsicsResolver, + lambda_alias: Optional[LambdaAlias] = None, + ) -> List[Any]: """Generates and returns the resources associated with this function's events. :param model.lambda_.LambdaFunction lambda_function: generated Lambda function @@ -761,7 +774,7 @@ def _generate_event_resources( # type: ignore[no-untyped-def] return resources - def _construct_code_dict(self): # type: ignore[no-untyped-def] + def _construct_code_dict(self) -> Dict[str, Any]: """Constructs Lambda Code Dictionary based on the accepted SAM artifact properties such as `InlineCode`, `CodeUri` and `ImageUri` and also raises errors if more than one of them is defined. `PackageType` determines which artifacts are considered. @@ -782,11 +795,11 @@ def _construct_code_dict(self): # type: ignore[no-untyped-def] # Inline function for transformation of inline code. # It accepts arbitrary argumemnts, because the arguments do not matter for the result. - def _construct_inline_code(*args, **kwargs): # type: ignore[no-untyped-def] + def _construct_inline_code(*args: Any, **kwargs: Dict[str, Any]) -> Dict[str, Any]: return {"ZipFile": self.InlineCode} # dispatch mechanism per artifact on how it needs to be transformed. - artifact_dispatch = { + artifact_dispatch: Dict[str, Callable[..., Dict[str, Any]]] = { "InlineCode": _construct_inline_code, "CodeUri": construct_s3_location_object, "ImageUri": construct_image_code_object, @@ -813,8 +826,8 @@ def _construct_inline_code(*args, **kwargs): # type: ignore[no-untyped-def] filtered_key = "ImageUri" else: raise InvalidResourceException(self.logical_id, "Either 'InlineCode' or 'CodeUri' must be set.") - dispatch_function = artifact_dispatch[filtered_key] - return dispatch_function(artifacts[filtered_key], self.logical_id, filtered_key) # type: ignore[operator] + dispatch_function: Callable[..., Dict[str, Any]] = artifact_dispatch[filtered_key] + return dispatch_function(artifacts[filtered_key], self.logical_id, filtered_key) def _construct_version( self, function: LambdaFunction, intrinsics_resolver: IntrinsicsResolver, code_sha256: Optional[str] = None @@ -969,7 +982,7 @@ def _validate_deployment_preference_and_add_update_policy( def _resolve_property_to_boolean( self, - property_value: Union[bool, str, dict], # type: ignore[type-arg] + property_value: Union[bool, str, Dict[str, Any]], property_name: str, intrinsics_resolver: IntrinsicsResolver, mappings_resolver: IntrinsicsResolver, @@ -1557,12 +1570,12 @@ def to_cloudformation(self, **kwargs): # type: ignore[no-untyped-def] resources = [] # Append any CFN resources: - intrinsics_resolver = kwargs["intrinsics_resolver"] - resources.append(self._construct_lambda_layer(intrinsics_resolver)) # type: ignore[no-untyped-call] + intrinsics_resolver: IntrinsicsResolver = kwargs["intrinsics_resolver"] + resources.append(self._construct_lambda_layer(intrinsics_resolver)) return resources - def _construct_lambda_layer(self, intrinsics_resolver): # type: ignore[no-untyped-def] + def _construct_lambda_layer(self, intrinsics_resolver: IntrinsicsResolver) -> LambdaLayerVersion: """Constructs and returns the Lambda function. :returns: a list containing the Lambda function and execution role resources @@ -1588,12 +1601,12 @@ def _construct_lambda_layer(self, intrinsics_resolver): # type: ignore[no-untyp old_logical_id = self.logical_id # This is to prevent the passthrough resource attributes to be included for hashing - hash_dict = copy.deepcopy(self.to_dict()) # type: ignore[no-untyped-call] - if "DeletionPolicy" in hash_dict.get(old_logical_id): + hash_dict = copy.deepcopy(self.to_dict()) + if "DeletionPolicy" in hash_dict.get(old_logical_id, {}): del hash_dict[old_logical_id]["DeletionPolicy"] - if "UpdateReplacePolicy" in hash_dict.get(old_logical_id): + if "UpdateReplacePolicy" in hash_dict.get(old_logical_id, {}): del hash_dict[old_logical_id]["UpdateReplacePolicy"] - if "Metadata" in hash_dict.get(old_logical_id): + if "Metadata" in hash_dict.get(old_logical_id, {}): del hash_dict[old_logical_id]["Metadata"] new_logical_id = logical_id_generator.LogicalIdGenerator(old_logical_id, hash_dict).gen() @@ -1615,10 +1628,10 @@ def _construct_lambda_layer(self, intrinsics_resolver): # type: ignore[no-untyp lambda_layer.LayerName = self.LayerName lambda_layer.Description = self.Description - lambda_layer.Content = construct_s3_location_object(self.ContentUri, self.logical_id, "ContentUri") # type: ignore[no-untyped-call] + lambda_layer.Content = construct_s3_location_object(self.ContentUri, self.logical_id, "ContentUri") lambda_layer.CompatibleArchitectures = self.CompatibleArchitectures - self._validate_architectures(lambda_layer) # type: ignore[no-untyped-call] + self._validate_architectures(lambda_layer) lambda_layer.CompatibleRuntimes = self.CompatibleRuntimes lambda_layer.LicenseInfo = self.LicenseInfo @@ -1658,7 +1671,7 @@ def _get_retention_policy_value(self) -> Optional[str]: "'RetentionPolicy' must be one of the following options: {}.".format([self.RETAIN, self.DELETE]), ) - def _validate_architectures(self, lambda_layer): # type: ignore[no-untyped-def] + def _validate_architectures(self, lambda_layer: LambdaLayerVersion) -> None: """Validate the values inside the CompatibleArchitectures field of a layer Parameters @@ -1718,7 +1731,7 @@ class SamStateMachine(SamResourceMacro): Tracing: Optional[Dict[str, Any]] PermissionsBoundary: Optional[Intrinsicable[str]] - event_resolver = ResourceTypeResolver( # type: ignore[no-untyped-call] + event_resolver = ResourceTypeResolver( samtranslator.model.stepfunctions.events, samtranslator.model.eventsources.scheduler, ) @@ -1756,13 +1769,13 @@ def to_cloudformation(self, **kwargs): # type: ignore[no-untyped-def] resources = state_machine_generator.to_cloudformation() return resources - def resources_to_link(self, resources): # type: ignore[no-untyped-def] + def resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]: try: - return {"event_resources": self._event_resources_to_link(resources)} # type: ignore[no-untyped-call] + return {"event_resources": self._event_resources_to_link(resources)} except InvalidEventException as e: raise InvalidResourceException(self.logical_id, e.message) - def _event_resources_to_link(self, resources): # type: ignore[no-untyped-def] + def _event_resources_to_link(self, resources: Dict[str, Any]) -> Dict[str, Any]: event_resources = {} if self.Events: for logical_id, event_dict in self.Events.items(): diff --git a/samtranslator/translator/translator.py b/samtranslator/translator/translator.py index 624e646dc..46abe0811 100644 --- a/samtranslator/translator/translator.py +++ b/samtranslator/translator/translator.py @@ -123,7 +123,7 @@ def translate( self.sam_parser.parse(sam_template=sam_template, parameter_values=parameter_values, sam_plugins=sam_plugins) template = copy.deepcopy(sam_template) - macro_resolver = ResourceTypeResolver(sam_resources) # type: ignore[no-untyped-call] + macro_resolver = ResourceTypeResolver(sam_resources) intrinsics_resolver = IntrinsicsResolver(parameter_values) # ResourceResolver is used by connector, its "resources" will be @@ -197,7 +197,7 @@ def translate( for logical_id in deployment_preference_collection.enabled_logical_ids(): try: template["Resources"].update( - deployment_preference_collection.deployment_group(logical_id).to_dict() # type: ignore[no-untyped-call] + deployment_preference_collection.deployment_group(logical_id).to_dict() ) except InvalidResourceException as e: document_errors.append(e) # type: ignore[arg-type]