""" Code for all Package-able resources """ import logging import os import shutil from typing import Dict, Optional, Union, cast import jmespath from botocore.utils import set_value_from_jmespath from samcli.commands.package import exceptions from samcli.lib.package.ecr_uploader import ECRUploader from samcli.lib.package.s3_uploader import S3Uploader from samcli.lib.package.uploaders import Destination, Uploaders from samcli.lib.package.utils import ( copy_to_temp_dir, is_ecr_url, is_local_file, is_path_value_valid, is_s3_protocol_url, is_zip_file, resource_not_packageable, upload_local_artifacts, upload_local_image_artifacts, ) from samcli.lib.utils import graphql_api from samcli.lib.utils.packagetype import IMAGE, ZIP from samcli.lib.utils.resources import ( AWS_APIGATEWAY_RESTAPI, AWS_APIGATEWAY_V2_API, AWS_APPSYNC_FUNCTIONCONFIGURATION, AWS_APPSYNC_GRAPHQLSCHEMA, AWS_APPSYNC_RESOLVER, AWS_CLOUDFORMATION_MODULEVERSION, AWS_CLOUDFORMATION_RESOURCEVERSION, AWS_ECR_REPOSITORY, AWS_ELASTICBEANSTALK_APPLICATIONVERSION, AWS_GLUE_JOB, AWS_LAMBDA_FUNCTION, AWS_LAMBDA_LAYERVERSION, AWS_SERVERLESS_API, AWS_SERVERLESS_FUNCTION, AWS_SERVERLESS_GRAPHQLAPI, AWS_SERVERLESS_HTTPAPI, AWS_SERVERLESS_LAYERVERSION, AWS_SERVERLESS_STATEMACHINE, AWS_SERVERLESSREPO_APPLICATION, AWS_STEPFUNCTIONS_STATEMACHINE, METADATA_WITH_LOCAL_PATHS, RESOURCES_WITH_IMAGE_COMPONENT, RESOURCES_WITH_LOCAL_PATHS, ) from samcli.lib.utils.s3 import parse_s3_url LOG = logging.getLogger(__name__) class Resource: RESOURCE_TYPE: Optional[str] = None PROPERTY_NAME: Optional[str] = None PACKAGE_NULL_PROPERTY = True # Set this property to True in base class if you want the exporter to zip # up the file before uploading This is useful for Lambda functions. FORCE_ZIP = False EXPORT_DESTINATION: Destination ARTIFACT_TYPE: Optional[str] = None def __init__(self, uploaders: Uploaders, code_signer, cache: Optional[Dict] = None): self.uploaders = uploaders self.code_signer = code_signer self.cache = cache @property def uploader(self) -> Union[S3Uploader, ECRUploader]: """ Return the uploader matching the EXPORT_DESTINATION """ return self.uploaders.get(self.EXPORT_DESTINATION) def export(self, resource_id, resource_dict, parent_dir): self.do_export(resource_id, resource_dict, parent_dir) def do_export(self, resource_id, resource_dict, parent_dir): pass def delete(self, resource_id, resource_dict): pass class ResourceZip(Resource): """ Base class representing a CloudFormation resource that can be exported """ RESOURCE_TYPE: str = "" PROPERTY_NAME: str = "" PACKAGE_NULL_PROPERTY = True # Set this property to True in base class if you want the exporter to zip # up the file before uploading This is useful for Lambda functions. FORCE_ZIP = False ARTIFACT_TYPE = ZIP EXPORT_DESTINATION = Destination.S3 def export(self, resource_id: str, resource_dict: Optional[Dict], parent_dir: str): if resource_dict is None: return if resource_not_packageable(resource_dict): return property_value = jmespath.search(self.PROPERTY_NAME, resource_dict) if not property_value and not self.PACKAGE_NULL_PROPERTY: return if isinstance(property_value, dict): LOG.debug("Property %s of %s resource is not a URL", self.PROPERTY_NAME, resource_id) return # If property is a file but not a zip file, place file in temp # folder and send the temp folder to be zipped temp_dir = None if is_local_file(property_value) and not is_zip_file(property_value) and self.FORCE_ZIP: temp_dir = copy_to_temp_dir(property_value) set_value_from_jmespath(resource_dict, self.PROPERTY_NAME, temp_dir) try: self.do_export(resource_id, resource_dict, parent_dir) except Exception as ex: LOG.debug("Unable to export", exc_info=ex) raise exceptions.ExportFailedError( resource_id=resource_id, property_name=self.PROPERTY_NAME, property_value=property_value, ex=ex ) finally: if temp_dir: shutil.rmtree(temp_dir) def do_export( self, resource_id, resource_dict, parent_dir, property_path: Optional[str] = None, local_path: Optional[str] = None, ): """ Default export action is to upload artifacts and set the property to S3 URL of the uploaded object If code signing configuration is provided for function/layer, uploaded artifact will be replaced by signed artifact location """ if property_path is None: property_path = self.PROPERTY_NAME uploader = cast(S3Uploader, self.uploader) # code signer only accepts files which has '.zip' extension in it # so package artifact with '.zip' if it is required to be signed should_sign_package = self.code_signer.should_sign_package(resource_id) artifact_extension = "zip" if should_sign_package else None uploaded_url = upload_local_artifacts( self.RESOURCE_TYPE, resource_id, resource_dict, property_path, parent_dir, uploader, artifact_extension, local_path, self.cache, ) if should_sign_package: uploaded_url = self.code_signer.sign_package( resource_id, uploaded_url, uploader.get_version_of_artifact(uploaded_url) ) set_value_from_jmespath(resource_dict, property_path, uploaded_url) def delete(self, resource_id, resource_dict): """ Delete the S3 artifact using S3 url referenced by PROPERTY_NAME """ if resource_dict is None: return s3_info = self.get_property_value(resource_dict) if s3_info["Key"]: self.uploader.delete_artifact(s3_info["Key"], True) def get_property_value(self, resource_dict): """ Get the s3 property value for this resource """ if resource_dict is None: return {"Bucket": None, "Key": None} resource_path = jmespath.search(self.PROPERTY_NAME, resource_dict) # In the case where resource_path is pointing to an intrinsinc # ref function, sam delete will delete the stack but skip the deletion of this # artifact, as deletion of intrinsic ref function artifacts is not supported yet. # TODO: Allow deletion of S3 artifacts with intrinsic ref functions. if resource_path and isinstance(resource_path, str): return parse_s3_url(resource_path) return {"Bucket": None, "Key": None} class ResourceImageDict(Resource): """ Base class representing a CFN Image based resource that can be exported. """ RESOURCE_TYPE: Optional[str] = None PROPERTY_NAME: Optional[str] = None FORCE_ZIP = False ARTIFACT_TYPE = IMAGE EXPORT_DESTINATION = Destination.ECR EXPORT_PROPERTY_CODE_KEY = "ImageUri" def export(self, resource_id, resource_dict, parent_dir): if resource_dict is None: return property_value = jmespath.search(self.PROPERTY_NAME, resource_dict) if isinstance(property_value, dict): LOG.debug("Property %s of %s resource is not a URL or a local image", self.PROPERTY_NAME, resource_id) return try: self.do_export(resource_id, resource_dict, parent_dir) except Exception as ex: LOG.debug("Unable to export", exc_info=ex) raise exceptions.ExportFailedError( resource_id=resource_id, property_name=self.PROPERTY_NAME, property_value=property_value, ex=ex ) def do_export(self, resource_id, resource_dict, parent_dir): """ Default export action is to upload artifacts and set the property to dictionary where the key is EXPORT_PROPERTY_CODE_KEY and value is set to an uploaded URL. """ uploaded_url = upload_local_image_artifacts( resource_id, resource_dict, self.PROPERTY_NAME, parent_dir, self.uploader ) set_value_from_jmespath(resource_dict, self.PROPERTY_NAME, {self.EXPORT_PROPERTY_CODE_KEY: uploaded_url}) def delete(self, resource_id, resource_dict): """ Delete the ECR artifact using ECR url in PROPERTY_NAME referenced by EXPORT_PROPERTY_CODE_KEY """ if resource_dict is None: return remote_path = resource_dict.get(self.PROPERTY_NAME, {}).get(self.EXPORT_PROPERTY_CODE_KEY) # In the case where remote_path is pointing to an intrinsinc # ref function, sam delete will delete the stack but skip the deletion of this # artifact, as deletion of intrinsic ref function artifacts is not supported yet. # TODO: Allow deletion of ECR artifacts with intrinsic ref functions. if isinstance(remote_path, str) and is_ecr_url(remote_path): self.uploader.delete_artifact( image_uri=remote_path, resource_id=resource_id, property_name=self.PROPERTY_NAME ) class ResourceImage(Resource): """ Base class representing a SAM Image based resource that can be exported. """ RESOURCE_TYPE: Optional[str] = None PROPERTY_NAME: Optional[str] = None FORCE_ZIP = False ARTIFACT_TYPE: Optional[str] = IMAGE EXPORT_DESTINATION = Destination.ECR def export(self, resource_id, resource_dict, parent_dir): if resource_dict is None: return property_value = jmespath.search(self.PROPERTY_NAME, resource_dict) if isinstance(property_value, dict): LOG.debug("Property %s of %s resource is not a URL or a local image", self.PROPERTY_NAME, resource_id) return try: self.do_export(resource_id, resource_dict, parent_dir) except Exception as ex: LOG.debug("Unable to export", exc_info=ex) raise exceptions.ExportFailedError( resource_id=resource_id, property_name=self.PROPERTY_NAME, property_value=property_value, ex=ex ) def do_export(self, resource_id, resource_dict, parent_dir): """ Default export action is to upload artifacts and set the property to URL of the uploaded object """ uploaded_url = upload_local_image_artifacts( resource_id, resource_dict, self.PROPERTY_NAME, parent_dir, self.uploader ) set_value_from_jmespath(resource_dict, self.PROPERTY_NAME, uploaded_url) def delete(self, resource_id, resource_dict): """ Delete the ECR artifact using ECR url referenced by property_name """ if resource_dict is None: return remote_path = resource_dict.get(self.PROPERTY_NAME) # In the case where remote_path is pointing to an intrinsinc # ref function, sam delete will delete the stack but skip the deletion of this # artifact, as deletion of intrinsic ref function artifacts is not supported yet. # TODO: Allow deletion of ECR artifacts with intrinsic ref functions. if isinstance(remote_path, str) and is_ecr_url(remote_path): self.uploader.delete_artifact( image_uri=remote_path, resource_id=resource_id, property_name=self.PROPERTY_NAME ) class ResourceWithS3UrlDict(ResourceZip): """ Represents CloudFormation resources that need the S3 URL to be specified as an dict like {Bucket: "", Key: "", Version: ""} """ BUCKET_NAME_PROPERTY: Optional[str] = None OBJECT_KEY_PROPERTY: Optional[str] = None VERSION_PROPERTY: Optional[str] = None ARTIFACT_TYPE = ZIP EXPORT_DESTINATION = Destination.S3 def do_export(self, resource_id, resource_dict, parent_dir): """ Upload to S3 and set property to an dict representing the S3 url of the uploaded object """ artifact_s3_url = upload_local_artifacts( self.RESOURCE_TYPE, resource_id, resource_dict, self.PROPERTY_NAME, parent_dir, self.uploader ) parsed_url = parse_s3_url( artifact_s3_url, bucket_name_property=self.BUCKET_NAME_PROPERTY, object_key_property=self.OBJECT_KEY_PROPERTY, version_property=self.VERSION_PROPERTY, ) set_value_from_jmespath(resource_dict, self.PROPERTY_NAME, parsed_url) def delete(self, resource_id, resource_dict): """ Delete the S3 artifact using S3 url in the dict PROPERTY_NAME using the bucket at BUCKET_NAME_PROPERTY and key at OBJECT_KEY_PROPERTY """ if resource_dict is None: return s3_info = self.get_property_value(resource_dict) if s3_info["Key"]: self.uploader.delete_artifact(remote_path=s3_info["Key"], is_key=True) def get_property_value(self, resource_dict): """ Get the s3 property value for this resource """ if resource_dict is None: return {"Bucket": None, "Key": None} resource_path = resource_dict.get(self.PROPERTY_NAME, {}) s3_bucket = resource_path.get(self.BUCKET_NAME_PROPERTY, None) key = resource_path.get(self.OBJECT_KEY_PROPERTY, None) # In the case where resource_path is pointing to an intrinsinc # ref function, sam delete will delete the stack but skip the deletion of this # artifact, as deletion of intrinsic ref function artifacts is not supported yet. # TODO: Allow deletion of S3 artifacts with intrinsic ref functions. if isinstance(s3_bucket, str) and isinstance(key, str): return {"Bucket": s3_bucket, "Key": key} return {"Bucket": None, "Key": None} class ServerlessFunctionResource(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESS_FUNCTION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] FORCE_ZIP = True class ServerlessFunctionImageResource(ResourceImage): RESOURCE_TYPE = AWS_SERVERLESS_FUNCTION PROPERTY_NAME = RESOURCES_WITH_IMAGE_COMPONENT[RESOURCE_TYPE][0] FORCE_ZIP = False class ServerlessApiResource(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESS_API PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] # Don't package the directory if DefinitionUri is omitted. # Necessary to support DefinitionBody PACKAGE_NULL_PROPERTY = False class ServerlessHttpApiResource(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESS_HTTPAPI PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] # Don't package the directory if DefinitionUri is omitted. # Necessary to support DefinitionBody PACKAGE_NULL_PROPERTY = False class ServerlessStateMachineResource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_SERVERLESS_STATEMACHINE PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] PACKAGE_NULL_PROPERTY = False BUCKET_NAME_PROPERTY = "Bucket" OBJECT_KEY_PROPERTY = "Key" VERSION_PROPERTY = "Version" class GraphQLSchemaResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_GRAPHQLSCHEMA PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] # Don't package the directory if DefinitionS3Location is omitted. # Necessary to support Definition PACKAGE_NULL_PROPERTY = False class AppSyncResolverRequestTemplateResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_RESOLVER PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] # Don't package the directory if RequestMappingTemplateS3Location is omitted. # Necessary to support RequestMappingTemplate PACKAGE_NULL_PROPERTY = False class AppSyncResolverResponseTemplateResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_RESOLVER PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][1] # Don't package the directory if ResponseMappingTemplateS3Location is omitted. # Necessary to support ResponseMappingTemplate PACKAGE_NULL_PROPERTY = False class AppSyncResolverCodeResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_RESOLVER PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][2] # Don't package the directory if CodeS3Location is omitted. # Necessary to support CodeS3Location PACKAGE_NULL_PROPERTY = False class AppSyncFunctionConfigurationRequestTemplateResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_FUNCTIONCONFIGURATION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] # Don't package the directory if RequestMappingTemplateS3Location is omitted. # Necessary to support RequestMappingTemplate PACKAGE_NULL_PROPERTY = False class AppSyncFunctionConfigurationResponseTemplateResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_FUNCTIONCONFIGURATION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][1] # Don't package the directory if ResponseMappingTemplateS3Location is omitted. # Necessary to support ResponseMappingTemplate PACKAGE_NULL_PROPERTY = False class AppSyncFunctionConfigurationCodeResource(ResourceZip): RESOURCE_TYPE = AWS_APPSYNC_FUNCTIONCONFIGURATION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][2] # Don't package the directory if CodeS3Location is omitted. # Necessary to support CodeS3Location PACKAGE_NULL_PROPERTY = False class LambdaFunctionResource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_LAMBDA_FUNCTION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] BUCKET_NAME_PROPERTY = "S3Bucket" OBJECT_KEY_PROPERTY = "S3Key" VERSION_PROPERTY = "S3ObjectVersion" FORCE_ZIP = True class LambdaFunctionImageResource(ResourceImage): RESOURCE_TYPE = AWS_LAMBDA_FUNCTION PROPERTY_NAME = RESOURCES_WITH_IMAGE_COMPONENT[RESOURCE_TYPE][0] FORCE_ZIP = True class StepFunctionsStateMachineResource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_STEPFUNCTIONS_STATEMACHINE PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] PACKAGE_NULL_PROPERTY = False BUCKET_NAME_PROPERTY = "Bucket" OBJECT_KEY_PROPERTY = "Key" VERSION_PROPERTY = "Version" class ApiGatewayV2Resource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_APIGATEWAY_V2_API PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] PACKAGE_NULL_PROPERTY = False BUCKET_NAME_PROPERTY = "Bucket" OBJECT_KEY_PROPERTY = "Key" VERSION_PROPERTY = "Version" class ApiGatewayRestApiResource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_APIGATEWAY_RESTAPI PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] PACKAGE_NULL_PROPERTY = False BUCKET_NAME_PROPERTY = "Bucket" OBJECT_KEY_PROPERTY = "Key" VERSION_PROPERTY = "Version" class ElasticBeanstalkApplicationVersion(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_ELASTICBEANSTALK_APPLICATIONVERSION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] BUCKET_NAME_PROPERTY = "S3Bucket" OBJECT_KEY_PROPERTY = "S3Key" VERSION_PROPERTY = None class LambdaLayerVersionResource(ResourceWithS3UrlDict): RESOURCE_TYPE = AWS_LAMBDA_LAYERVERSION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] BUCKET_NAME_PROPERTY = "S3Bucket" OBJECT_KEY_PROPERTY = "S3Key" VERSION_PROPERTY = "S3ObjectVersion" FORCE_ZIP = True class ServerlessLayerVersionResource(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESS_LAYERVERSION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] FORCE_ZIP = True class ServerlessRepoApplicationLicense(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESSREPO_APPLICATION PROPERTY_NAME = METADATA_WITH_LOCAL_PATHS[RESOURCE_TYPE][0] PACKAGE_NULL_PROPERTY = False class ServerlessRepoApplicationReadme(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESSREPO_APPLICATION PROPERTY_NAME = METADATA_WITH_LOCAL_PATHS[RESOURCE_TYPE][1] PACKAGE_NULL_PROPERTY = False class GlueJobCommandScriptLocationResource(ResourceZip): """ Represents Glue::Job resource. """ RESOURCE_TYPE = AWS_GLUE_JOB # Note the PROPERTY_NAME includes a '.' implying it's nested. PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[AWS_GLUE_JOB][0] class CloudFormationModuleVersionModulePackage(ResourceZip): RESOURCE_TYPE = AWS_CLOUDFORMATION_MODULEVERSION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[AWS_CLOUDFORMATION_MODULEVERSION][0] class CloudFormationResourceVersionSchemaHandlerPackage(ResourceZip): RESOURCE_TYPE = AWS_CLOUDFORMATION_RESOURCEVERSION PROPERTY_NAME = RESOURCES_WITH_LOCAL_PATHS[AWS_CLOUDFORMATION_RESOURCEVERSION][0] class ECRResource(Resource): """ Represents CloudFormation resources ECR for deleting the ECR repository with the property name RepositoryName. This class is used only for deleting the repository and not exporting anything. """ RESOURCE_TYPE = AWS_ECR_REPOSITORY PROPERTY_NAME = RESOURCES_WITH_IMAGE_COMPONENT[RESOURCE_TYPE][0] ARTIFACT_TYPE = ZIP EXPORT_DESTINATION = Destination.ECR def delete(self, resource_id, resource_dict): if resource_dict is None: return repository_name = self.get_property_value(resource_dict) # TODO: Allow deletion of ECR Repositories with intrinsic ref functions. if repository_name and isinstance(repository_name, str): self.uploader.delete_ecr_repository(physical_id=repository_name) def get_property_value(self, resource_dict): if resource_dict is None: return None return jmespath.search(self.PROPERTY_NAME, resource_dict) class GraphQLApiSchemaResource(ResourceZip): RESOURCE_TYPE = AWS_SERVERLESS_GRAPHQLAPI PROPERTY_NAME = graphql_api.SCHEMA_ARTIFACT_PROPERTY # Don't package the directory if SchemaUri is omitted. # Necessary to support SchemaInline PACKAGE_NULL_PROPERTY = False class GraphQLApiCodeResource(ResourceZip): """CodeUri for GraphQLApi resource. There can be more than a single instance of CodeUri property in GraphQLApi Resolvers and Functions. This class handles them all. GraphQLApi dict shape looks like the following (yaml representation) >>> Resolvers: Mutation: Resolver1: CodeUri: ... Pipeline: - Func1 - Func2 Query: Resolver2: CodeUri: ... Pipeline: - Func3 Functions: Func1: CodeUri: ... Func2: CodeUri: ... Func3: CodeUri: ... ... # other properties, which are not important here """ RESOURCE_TYPE = AWS_SERVERLESS_GRAPHQLAPI PROPERTY_NAME = graphql_api.CODE_ARTIFACT_PROPERTY # if CodeUri is omitted the directory is not packaged because it's necessary to support CodeInline PACKAGE_NULL_PROPERTY = False def export(self, resource_id: str, resource_dict: Optional[Dict], parent_dir: str): if resource_dict is None: return if resource_not_packageable(resource_dict): return # to be able to set different nested properties to S3 uri, paths are necessary # jmespath doesn't provide that functionality, thus custom implementation paths_values = graphql_api.find_all_paths_and_values(self.PROPERTY_NAME, resource_dict) for property_path, property_value in paths_values: if isinstance(property_value, dict): LOG.debug("Property %s of %s resource is not a URL", self.PROPERTY_NAME, resource_id) return # If property is a file but not a zip file, place file in temp # folder and send the temp folder to be zipped temp_dir = None if is_local_file(property_value) and not is_zip_file(property_value) and self.FORCE_ZIP: temp_dir = copy_to_temp_dir(property_value) set_value_from_jmespath(resource_dict, property_path, temp_dir) try: self.do_export( resource_id, resource_dict, parent_dir, property_path=property_path, local_path=property_value ) except Exception as ex: LOG.debug("Unable to export", exc_info=ex) raise exceptions.ExportFailedError( resource_id=resource_id, property_name=property_path, property_value=property_value, ex=ex ) finally: if temp_dir: shutil.rmtree(temp_dir) RESOURCES_EXPORT_LIST = [ ServerlessFunctionResource, ServerlessFunctionImageResource, ServerlessApiResource, ServerlessHttpApiResource, ServerlessStateMachineResource, GraphQLSchemaResource, AppSyncResolverRequestTemplateResource, AppSyncResolverResponseTemplateResource, AppSyncResolverCodeResource, AppSyncFunctionConfigurationRequestTemplateResource, AppSyncFunctionConfigurationResponseTemplateResource, AppSyncFunctionConfigurationCodeResource, ApiGatewayRestApiResource, ApiGatewayV2Resource, StepFunctionsStateMachineResource, LambdaFunctionResource, LambdaFunctionImageResource, ElasticBeanstalkApplicationVersion, ServerlessLayerVersionResource, LambdaLayerVersionResource, GlueJobCommandScriptLocationResource, CloudFormationModuleVersionModulePackage, CloudFormationResourceVersionSchemaHandlerPackage, ECRResource, GraphQLApiSchemaResource, GraphQLApiCodeResource, ] METADATA_EXPORT_LIST = [ServerlessRepoApplicationReadme, ServerlessRepoApplicationLicense] def include_transform_export_handler(template_dict, uploader, parent_dir): if template_dict.get("Name", None) != "AWS::Include": return template_dict include_location = template_dict.get("Parameters", {}).get("Location", None) if not include_location or not is_path_value_valid(include_location) or is_s3_protocol_url(include_location): # `include_location` is either empty, or not a string, or an S3 URI return template_dict # We are confident at this point that `include_location` is a string containing the local path abs_include_location = os.path.join(parent_dir, include_location) if is_local_file(abs_include_location): template_dict["Parameters"]["Location"] = uploader.upload_with_dedup(abs_include_location) else: raise exceptions.InvalidLocalPathError( resource_id="AWS::Include", property_name="Location", local_path=abs_include_location ) return template_dict GLOBAL_EXPORT_DICT = {"Fn::Transform": include_transform_export_handler}