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