forked from aws-powertools/powertools-lambda-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlayer_stack.py
More file actions
175 lines (150 loc) · 5.3 KB
/
layer_stack.py
File metadata and controls
175 lines (150 loc) · 5.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
from typing import Optional
import jsii
from aws_cdk import (
Aspects,
CfnCondition,
CfnOutput,
CfnParameter,
CfnResource,
Fn,
IAspect,
RemovalPolicy,
Stack,
)
from aws_cdk.aws_lambda import Architecture, CfnLayerVersionPermission
from aws_cdk.aws_ssm import StringParameter
from cdk_aws_lambda_powertools_layer import LambdaPowertoolsLayer
from constructs import Construct
@jsii.implements(IAspect)
class ApplyCondition:
def __init__(self, condition: CfnCondition):
self.condition = condition
def visit(self, node):
if isinstance(node, CfnResource):
node.cfn_options.condition = self.condition
if isinstance(node, CfnOutput):
node.condition = self.condition
class Layer(Construct):
def __init__(
self,
scope: Construct,
construct_id: str,
layer_version_name: str,
powertools_version: str,
architecture: Optional[Architecture] = None,
**kwargs
) -> None:
super().__init__(scope, construct_id, **kwargs)
layer = LambdaPowertoolsLayer(
self,
"Layer",
layer_version_name=layer_version_name,
version=powertools_version,
include_extras=True,
compatible_architectures=[architecture] if architecture else [],
)
layer.apply_removal_policy(RemovalPolicy.RETAIN)
self.layer_version_arn = layer.layer_version_arn
layer_permission = CfnLayerVersionPermission(
self,
"PublicLayerAccess",
action="lambda:GetLayerVersion",
layer_version_arn=layer.layer_version_arn,
principal="*",
)
layer_permission.apply_removal_policy(RemovalPolicy.RETAIN)
class LayerStack(Stack):
def __init__(
self,
scope: Construct,
construct_id: str,
powertools_version: str,
ssm_parameter_layer_arn: str,
ssm_parameter_layer_arm64_arn: str,
**kwargs
) -> None:
super().__init__(scope, construct_id, **kwargs)
has_arm64_support = CfnParameter(
self,
"HasARM64Support",
description="Has ARM64 Support Condition",
type="String",
allowed_values=["true", "false"],
)
has_arm64_condition = CfnCondition(
self,
"HasARM64SupportCondition",
expression=Fn.condition_equals(has_arm64_support, "true"),
)
has_no_arm64_condition = CfnCondition(
self,
"HasNOArm64SupportCondition",
expression=Fn.condition_equals(has_arm64_support, "false"),
)
# The following code is used when the region does not support ARM64 Lambdas. We make sure to only create the
# X86_64 Layer without specifying any compatible architecture, which would result in a CloudFormation error.
layer_single = Layer(
self,
"LayerSingle",
layer_version_name="AWSLambdaPowertoolsPythonV2",
powertools_version=powertools_version,
)
Aspects.of(layer_single).add(ApplyCondition(has_no_arm64_condition))
Aspects.of(
StringParameter(
self,
"SingleVersionArn",
parameter_name=ssm_parameter_layer_arn,
string_value=layer_single.layer_version_arn,
)
).add(ApplyCondition(has_no_arm64_condition))
# The following code is used when the region has support for ARM64 Lambdas. In this case, we explicitly
# create a Layer for both X86_64 and ARM64, specifying the compatible architectures.
# X86_64 layer
layer = Layer(
self,
"Layer",
layer_version_name="AWSLambdaPowertoolsPythonV2",
powertools_version=powertools_version,
architecture=Architecture.X86_64,
)
Aspects.of(layer).add(ApplyCondition(has_arm64_condition))
Aspects.of(
StringParameter(
self,
"VersionArn",
parameter_name=ssm_parameter_layer_arn,
string_value=layer.layer_version_arn,
)
).add(ApplyCondition(has_arm64_condition))
CfnOutput(
self,
"LatestLayerArn",
value=Fn.condition_if(
has_arm64_condition.logical_id,
layer.layer_version_arn,
layer_single.layer_version_arn,
).to_string(),
)
# ARM64 layer
layer_arm64 = Layer(
self,
"Layer-ARM64",
layer_version_name="AWSLambdaPowertoolsPythonV2-Arm64",
powertools_version=powertools_version,
architecture=Architecture.ARM_64,
)
Aspects.of(layer_arm64).add(ApplyCondition(has_arm64_condition))
StringParameter(
self,
"Arm64VersionArn",
parameter_name=ssm_parameter_layer_arm64_arn,
string_value=Fn.condition_if(
has_arm64_condition.logical_id,
layer_arm64.layer_version_arn,
"none",
).to_string(),
)
Aspects.of(
CfnOutput(self, "LatestLayerArm64Arn", value=layer_arm64.layer_version_arn)
).add(ApplyCondition(has_arm64_condition))