Untitled
unknown
plain_text
6 months ago
12 kB
4
Indexable
import json import os import aws_cdk from aws_cdk import Duration, Stack, aws_iam, aws_lambda, aws_logs from principal_environment import PrincipalEnvironment from aws_constructs.iam_role import IAMRoleConstruct from aws_constructs.lambda_triggers import LambdaTriggers EXCLUDE_LIST = [ "layer", "custom_layer", "python.zip", "tests", "test", "coverage", "__init__.py", "requirements_dev.txt", "requirements.txt", "config.json", ] IMG_EXCLUDE_LIST = EXCLUDE_LIST.copy() IMG_EXCLUDE_LIST.remove("requirements.txt") class LambdaConstruct: @staticmethod def create_lambda( scope: Stack, app_name: str, code_dir: str, env: PrincipalEnvironment, identifier_name: str = "", ) -> None: print(f"Deploying Lambda From {identifier_name}/{code_dir}..........") if os.path.isfile( f"../../src/lambdas/{identifier_name}/{code_dir}/config.json" ): config_path = f"../../src/lambdas/{identifier_name}/{code_dir}/config.json" else: config_path = "../../src/lambdas/default_lambda_config.json" with open(config_path, "r", encoding="utf8") as file: lmd_config_json = json.load(file)[env.aws_environment_name] env_vars = { "function_name": f"{app_name}-{code_dir}-lmd".replace("_", "-"), "env": env.aws_environment_name, "region": env.region, "account": env.account, "app_name": app_name, } if lmd_config_json["lambda_prop"].get("env_vars"): env_vars.update(lmd_config_json["lambda_prop"]["env_vars"]) lmd_props = { "architecture": aws_lambda.Architecture.X86_64, "timeout": Duration.minutes(lmd_config_json["lambda_prop"]["timeout"]), "environment": LambdaConstruct.get_env_args(scope, env, env_vars), "memory_size": lmd_config_json["lambda_prop"]["memory_size"], "ephemeral_storage_size": aws_cdk.Size.mebibytes( lmd_config_json["lambda_prop"]["ephemeral_storage_size"] ), "retry_attempts": lmd_config_json["lambda_prop"].get("retry_attempts", 2), } if lmd_config_json["lambda_prop"].get("add_vpc"): vpc_config = aws_cdk.aws_ec2.Vpc.from_vpc_attributes( scope, f"{code_dir}-vpc-config", availability_zones=scope.cdk[env.aws_environment_name]["az"][ env.region ], vpc_id=scope.cdk[env.aws_environment_name]["vpc_id"][env.region], private_subnet_ids=scope.cdk[env.aws_environment_name]["subnets"][ env.region ], ) lmd_props.update( { "vpc": vpc_config, "security_groups": [ aws_cdk.aws_ec2.SecurityGroup.from_lookup_by_name( scope, f"LambdaSG-{code_dir}", "prinam-hk-fa-lmd-sg", vpc_config, ) ], } ) fn_obj = aws_lambda.Function( scope, f"{code_dir}-lmd".replace("_", "-"), code=aws_lambda.Code.from_asset( path=f"../../src/lambdas/{identifier_name}/{code_dir}", exclude=EXCLUDE_LIST, ), function_name=f"{app_name}-{code_dir}-lmd".replace("_", "-"), handler=lmd_config_json["lambda_prop"]["handler"], role=IAMRoleConstruct.create_lambda_role( scope, app_name, code_dir.replace("_", "-"), env, lmd_config_json.get("add_policy", []), ), runtime=aws_lambda.Runtime( lmd_config_json["lambda_prop"]["runtime"], family=aws_lambda.RuntimeFamily.PYTHON, ), # layers=LambdaConstruct.create_all_layers( # scope, # f"{app_name}-{code_dir}-lyr".replace("_", "-"), # app_name, # identifier_name, # code_dir, # config_path, # lmd_config_json["lambda_prop"]["runtime"], # env, # ), **lmd_props, ) LambdaConstruct.configure_lmd(scope, lmd_config_json, fn_obj, env, app_name, code_dir) @staticmethod def configure_lmd( scope: Stack, config_json: dict, lmd_obj: aws_lambda.Function, env: PrincipalEnvironment, app_name: str, code_dir: str, ) -> None: if config_json.get("lambda_triggers"): LambdaTriggers.add_lambda_trigger( scope, lmd_obj, env, config_json["lambda_triggers"], f"{app_name}-{code_dir}".replace("_", "-"), ) aws_logs.LogRetention( scope, f"LogRetention-{code_dir}", log_group_name=f"/aws/lambda/{lmd_obj.function_name}", retention=aws_logs.RetentionDays.THREE_MONTHS, removal_policy=aws_cdk.RemovalPolicy.DESTROY, role=aws_iam.Role.from_role_name( scope, f"CustomResourceRole-{code_dir}-{env.region}", f"{app_name}-{code_dir.replace('_', '-')}-lmd-role", ), ) # @staticmethod # def create_all_layers( # scope: Stack, # layer_id: str, # app_name: str, # id_name: str, # deps_dir: str, # conf_file_path: str, # runtime: str, # env: PrincipalEnvironment, # ) -> Optional[Sequence[ILayerVersion]]: # all_layers = [] # with open(conf_file_path, "r", encoding="utf8") as file: # lambda_layer_prop = json.load(file)[env.aws_environment_name][ # "lambda_layer_prop" # ] # if lambda_layer_prop.get("default_layer", True): # if lambda_layer_prop.get("default_layer") == "zip": # all_layers.append( # aws_lambda.LayerVersion( # scope, # f"def-lyr-{layer_id}", # layer_version_name=f"{app_name}-{deps_dir}-def-lyr".replace( # "_", "-" # ), # code=aws_lambda.Code.from_asset( # f"../../src/lambdas/layers/default_layer/" # f"awswrangler-layer-3.4.2-py{runtime.split('python')[1]}.zip" # ), # compatible_architectures=[aws_lambda.Architecture.X86_64], # compatible_runtimes=[ # aws_lambda.Runtime( # runtime, family=aws_lambda.RuntimeFamily.PYTHON # ) # ], # ) # ) # else: # all_layers.append( # aws_lambda.LayerVersion.from_layer_version_arn( # scope, # f"def-lyr-{layer_id}", # layer_version_arn=f"arn:aws:lambda:{env.region}:336392948345:layer:" # f"AWSSDKPandas-{scope.cdk['pandas_sdk'][runtime]}", # ) # ) # if lambda_layer_prop.get("custom_layer", False): # ctm_lib_name = lambda_layer_prop["custom_layer"].split("==")[0] # version = lambda_layer_prop["custom_layer"].split("==")[1] # all_layers.append( # aws_lambda.LayerVersion( # scope, # f"ctm-lyr-{layer_id}", # layer_version_name=f"{app_name}-{deps_dir}-{ctm_lib_name}-lyr".replace( # "_", "-" # ), # code=aws_lambda.Code.from_asset( # f"../../src/lambdas/layers/custom_layer/" # f"{ctm_lib_name}-layer-{version}-py{runtime.split('python')[1]}.zip" # ), # compatible_architectures=[aws_lambda.Architecture.X86_64], # compatible_runtimes=[ # aws_lambda.Runtime( # runtime, family=aws_lambda.RuntimeFamily.PYTHON # ) # ], # ) # ) # if lambda_layer_prop.get("data_reservoir_layer", False): # layer_location = "../../src/lambdas/layers/data_reservoir_layer" # if os.path.exists(layer_location): # shutil.rmtree(layer_location) # shutil.copytree( # "../../src/lambdas/layers/data_reservoir", # layer_location + "/python/data_reservoir", # ) # shutil.make_archive("data_reservoir", "zip", layer_location) # all_layers.append( # aws_lambda.LayerVersion( # scope, # f"data-reservoir-lyr-{layer_id}", # layer_version_name="prinam-my-fa-data-reservoir-lyr", # code=aws_lambda.Code.from_asset("data_reservoir.zip"), # compatible_architectures=[aws_lambda.Architecture.X86_64], # compatible_runtimes=[ # aws_lambda.Runtime( # runtime, family=aws_lambda.RuntimeFamily.PYTHON # ) # ], # ) # ) # if os.path.isfile(f"../../src/lambdas/{id_name}/{deps_dir}/requirements.txt"): # layer_location = ( # f"../../src/lambdas/{id_name}/{deps_dir}/layer/requirements" # ) # if not os.path.isdir(layer_location): # print(f"Layer dir doesn't exist in {deps_dir}, creating...") # pip.main( # [ # "install", # "-q", # "--only-binary=:all:", # "--platform", # "manylinux2014_x86_64", # "--python-version", # runtime.split("python")[1], # "-r", # f"../../src/lambdas/{id_name}/{deps_dir}/requirements.txt", # "-t", # f"{layer_location}/python", # ] # ) # all_layers.append( # aws_lambda.LayerVersion( # scope, # layer_id, # layer_version_name=layer_id, # code=aws_lambda.Code.from_asset(layer_location), # compatible_architectures=[aws_lambda.Architecture.X86_64], # compatible_runtimes=[ # aws_lambda.Runtime( # runtime, family=aws_lambda.RuntimeFamily.PYTHON # ) # ], # ) # ) # return all_layers @staticmethod def get_env_args(scope: Stack, env: PrincipalEnvironment, args: dict) -> dict: mapped_values = { "env": env.aws_environment_name, "account": env.account, "region": env.region, "app_name": scope.app_name, "source_name": scope.source_name, } json_str = json.dumps(args) for key, val in mapped_values.items(): placeholder = f"<{key}>" json_str = json_str.replace(placeholder, val) return json.loads(json_str)
Editor is loading...
Leave a Comment