Untitled
unknown
plain_text
5 months ago
12 kB
4
Indexable
import json
import os
import shutil
from typing import Optional, Sequence
import aws_cdk
import pip
from aws_cdk import Duration, Stack, aws_iam, aws_lambda, aws_logs
from aws_cdk.aws_lambda import ILayerVersion
from aws_constructs.iam_role import IAMRoleConstruct
from aws_constructs.lambda_triggers import LambdaTriggers
from principal_environment import PrincipalEnvironment
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