in apigw-canary-deployment-cdk/app.py [0:0]
def __init__(self, scope: cdk.Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Create a Lambda function
# Code in ./src directory
lambda_fn = lambda_.Function(
self, "MyFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="index.handler",
code=lambda_.Code.from_asset(os.path.join(DIRNAME, "src"))
)
# Function version
version = lambda_fn.current_version
# Create function alias
alias = lambda_.Alias(
self, "LambdaAlias",
alias_name="Prod",
version=version
)
# Create the Rest API
rest_api = apigateway.RestApi(
self, "RestApi",
endpoint_types=[apigateway.EndpointType.REGIONAL],
deploy=False,
retain_deployments=False
)
# Create initial deployment
deployment = apigateway.Deployment(
self,
"Deployment",
api=rest_api,
retain_deployments=False
)
# Create prod Stage
stage = apigateway.Stage(
self,
"prod",
deployment=deployment,
variables={
"lambdaAlias": "Prod"
}
)
rest_api.deployment_stage = stage
# Create URI for lambda alias
stage_uri = f"arn:aws:apigateway:{cdk.Aws.REGION}:lambda:path/2015-03-31/functions/{lambda_fn.function_arn}:${{stageVariables.lambdaAlias}}/invocations"
# Create Lambda Integration
integration = apigateway.Integration(
type=apigateway.IntegrationType.AWS_PROXY,
integration_http_method="POST",
uri=stage_uri
)
# Create APIGW Method
method = rest_api.root.add_method("GET", integration)
# Add Lambda permissions
lambda_fn.add_permission(
"lambdaPermission",
action="lambda:InvokeFunction",
principal=iam.ServicePrincipal("apigateway.amazonaws.com"),
source_arn=method.method_arn.replace(
rest_api.deployment_stage.stage_name, "*"
)
)
# Add permissions for Prod alias
alias.add_permission(
"aliasPermission",
action="lambda:InvokeFunction",
principal=iam.ServicePrincipal("apigateway.amazonaws.com"),
source_arn=method.method_arn.replace(
rest_api.deployment_stage.stage_name, "*"
)
)
# OUTPUTS
cdk.CfnOutput(self, "LambdaFunction", export_name="MyLambdaFunction", value=lambda_fn.function_arn)
cdk.CfnOutput(self, "ApigwId", export_name="MyAPIGWID", value=rest_api.rest_api_id)
cdk.CfnOutput(self, "MethodArn", export_name="MyMethodArn", value=method.method_arn)
cdk.CfnOutput(self, "StageName", export_name="MyStageName", value=rest_api.deployment_stage.stage_name)