Untitled

 avatar
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