Chapter 6: Creating Lambda Functions with CDK

Haiyue
27min

Chapter 6: Creating Lambda Functions with CDK

Chapter Overview

This chapter provides detailed instructions on how to create, configure, and deploy Lambda functions using Python CDK. We will start with a simple Hello World function and progressively dive into complex function configurations and deployment strategies.

Learning Objectives

  1. Master the complete process of creating Lambda functions with CDK
  2. Understand various Lambda function configuration options
  3. Learn to handle function code packaging and deployment
  4. Master environment variable and permission configuration
  5. Understand different deployment patterns and best practices

6.1 Lambda Function Basic Configuration

6.1.1 Simplest Lambda Function

# stacks/lambda_stack.py
from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    Duration
)
from constructs import Construct

class LambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create simplest Lambda function
        self.hello_function = _lambda.Function(
            self, "HelloWorldFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_inline("""
def handler(event, context):
    return {
        'statusCode': 200,
        'body': 'Hello from Lambda!'
    }
            """),
            function_name="hello-world-function"
        )

6.1.2 Creating Lambda Function from File

# Project structure
my_lambda_cdk/
├── lambda_functions/
│   └── hello_world/
│       ├── index.py
│       └── requirements.txt
└── stacks/
    └── lambda_stack.py
# lambda_functions/hello_world/index.py
import json
import os
from datetime import datetime

def handler(event, context):
    """
    Lambda function handler
    """
    response = {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps({
            'message': 'Hello from Lambda CDK!',
            'timestamp': datetime.utcnow().isoformat(),
            'function_name': context.function_name,
            'function_version': context.function_version,
            'environment': os.getenv('ENVIRONMENT', 'unknown')
        })
    }

    return response
# stacks/lambda_stack.py (updated version)
from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    Duration
)
from constructs import Construct
import os

class LambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create Lambda function from directory
        self.hello_function = _lambda.Function(
            self, "HelloWorldFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset(
                os.path.join(os.path.dirname(__file__),
                           "../lambda_functions/hello_world")
            ),
            function_name="hello-world-function",
            timeout=Duration.seconds(30),
            memory_size=128,
            environment={
                "ENVIRONMENT": "development",
                "LOG_LEVEL": "INFO"
            }
        )

6.2 Lambda Function Configuration Details

6.2.1 Runtime and Handler Configuration

from aws_cdk import aws_lambda as _lambda

class AdvancedLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Supported Python runtimes
        runtimes = {
            'python38': _lambda.Runtime.PYTHON_3_8,
            'python39': _lambda.Runtime.PYTHON_3_9,
            'python310': _lambda.Runtime.PYTHON_3_10,
            'python311': _lambda.Runtime.PYTHON_3_11
        }

        # Create multiple versions of function
        self.functions = {}
        for name, runtime in runtimes.items():
            self.functions[name] = _lambda.Function(
                self, f"Function{name.title()}",
                runtime=runtime,
                handler="index.handler",
                code=_lambda.Code.from_asset("lambda_functions/multi_runtime"),
                function_name=f"multi-runtime-{name}",
                description=f"Function using {runtime.name}"
            )

6.2.2 Memory and Timeout Configuration

from aws_cdk import Duration

class PerformanceOptimizedStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Function configurations for different performance needs
        function_configs = [
            {
                'name': 'LightweightFunction',
                'memory': 128,      # Minimum memory
                'timeout': 15,      # 15 second timeout
                'description': 'Lightweight processing function'
            },
            {
                'name': 'StandardFunction',
                'memory': 512,      # Standard memory
                'timeout': 60,      # 1 minute timeout
                'description': 'Standard processing function'
            },
            {
                'name': 'HeavyFunction',
                'memory': 3008,     # Maximum memory
                'timeout': 900,     # 15 minute timeout
                'description': 'Heavy computation function'
            }
        ]

        self.functions = {}
        for config in function_configs:
            self.functions[config['name']] = _lambda.Function(
                self, config['name'],
                runtime=_lambda.Runtime.PYTHON_3_9,
                handler="index.handler",
                code=_lambda.Code.from_asset("lambda_functions/performance"),
                memory_size=config['memory'],
                timeout=Duration.seconds(config['timeout']),
                description=config['description'],
                environment={
                    'MEMORY_SIZE': str(config['memory']),
                    'TIMEOUT': str(config['timeout'])
                }
            )

6.2.3 Environment Variable Configuration

import os
from aws_cdk import aws_ssm as ssm

class ConfigurableStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Get configuration from SSM Parameter Store
        api_key_param = ssm.StringParameter.from_string_parameter_name(
            self, "ApiKeyParam",
            string_parameter_name="/myapp/api-key"
        )

        # Complex environment variable configuration
        environment_vars = {
            # Basic configuration
            'ENVIRONMENT': os.getenv('ENVIRONMENT', 'dev'),
            'AWS_REGION': self.region,
            'LOG_LEVEL': 'DEBUG' if os.getenv('ENVIRONMENT') == 'dev' else 'INFO',

            # Feature flags
            'FEATURE_FLAG_NEW_API': 'true',
            'ENABLE_CACHE': 'true',
            'MAX_RETRY_ATTEMPTS': '3',

            # Service configuration
            'DATABASE_TIMEOUT': '30',
            'API_TIMEOUT': '60',
            'BATCH_SIZE': '100',

            # External services
            'EXTERNAL_API_ENDPOINT': 'https://api.example.com',
            'API_KEY_PARAM': api_key_param.parameter_name
        }

        self.configurable_function = _lambda.Function(
            self, "ConfigurableFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/configurable"),
            environment=environment_vars,
            function_name="configurable-function"
        )

        # Grant access to SSM parameter
        api_key_param.grant_read(self.configurable_function)

6.3 Code Packaging and Dependency Management

6.3.1 Handling Python Dependencies

# lambda_functions/with_dependencies/requirements.txt
requests==2.28.1
boto3==1.26.137
pandas==1.5.3
numpy==1.24.3
# lambda_functions/with_dependencies/index.py
import json
import requests
import pandas as pd
import boto3
from datetime import datetime

def handler(event, context):
    """
    Lambda function using third-party libraries
    """
    try:
        # Using requests library
        response = requests.get('https://httpbin.org/json', timeout=10)
        external_data = response.json()

        # Using pandas to process data
        df = pd.DataFrame([
            {'name': 'Alice', 'age': 25},
            {'name': 'Bob', 'age': 30},
            {'name': 'Charlie', 'age': 35}
        ])

        # Using boto3
        s3_client = boto3.client('s3')

        result = {
            'statusCode': 200,
            'body': json.dumps({
                'external_data': external_data,
                'processed_data': df.to_dict('records'),
                'timestamp': datetime.utcnow().isoformat()
            })
        }

    except Exception as e:
        result = {
            'statusCode': 500,
            'body': json.dumps({
                'error': str(e),
                'timestamp': datetime.utcnow().isoformat()
            })
        }

    return result
# stacks/lambda_with_deps_stack.py
from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    BundlingOptions,
    Duration
)
import os

class LambdaWithDepsStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Method 1: Use bundling to automatically install dependencies
        self.deps_function = _lambda.Function(
            self, "DependenciesFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset(
                "lambda_functions/with_dependencies",
                bundling=BundlingOptions(
                    image=_lambda.Runtime.PYTHON_3_9.bundling_image,
                    command=[
                        "bash", "-c",
                        "pip install -r requirements.txt -t /asset-output && cp -au . /asset-output"
                    ]
                )
            ),
            timeout=Duration.seconds(60),
            memory_size=256
        )

6.3.2 Using Docker Build

# lambda_functions/docker_function/Dockerfile
FROM public.ecr.aws/lambda/python:3.9

# Copy requirements and install dependencies
COPY requirements.txt ${LAMBDA_TASK_ROOT}
RUN pip install -r requirements.txt

# Copy function code
COPY index.py ${LAMBDA_TASK_ROOT}

# Set handler
CMD ["index.handler"]
# stacks/docker_lambda_stack.py
from aws_cdk import aws_lambda as _lambda

class DockerLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create Lambda function using Docker image
        self.docker_function = _lambda.Function(
            self, "DockerFunction",
            runtime=_lambda.Runtime.FROM_IMAGE,
            handler=_lambda.Handler.FROM_IMAGE,
            code=_lambda.Code.from_asset_image(
                "lambda_functions/docker_function"
            ),
            timeout=Duration.seconds(60),
            memory_size=512,
            function_name="docker-lambda-function"
        )

6.4 Lambda Function Permission Management

6.4.1 Basic IAM Permissions

from aws_cdk import (
    aws_iam as iam,
    aws_s3 as s3,
    aws_dynamodb as dynamodb
)

class LambdaPermissionsStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create S3 bucket
        self.bucket = s3.Bucket(
            self, "MyBucket",
            bucket_name="my-lambda-cdk-bucket"
        )

        # Create DynamoDB table
        self.table = dynamodb.Table(
            self, "MyTable",
            table_name="my-lambda-table",
            partition_key=dynamodb.Attribute(
                name="id",
                type=dynamodb.AttributeType.STRING
            )
        )

        # Create Lambda function
        self.lambda_function = _lambda.Function(
            self, "PermissionsFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/permissions"),
            environment={
                'BUCKET_NAME': self.bucket.bucket_name,
                'TABLE_NAME': self.table.table_name
            }
        )

        # Grant S3 permissions
        self.bucket.grant_read_write(self.lambda_function)

        # Grant DynamoDB permissions
        self.table.grant_read_write_data(self.lambda_function)

        # Add custom IAM policy
        custom_policy = iam.PolicyStatement(
            effect=iam.Effect.ALLOW,
            actions=[
                "secretsmanager:GetSecretValue",
                "ssm:GetParameter",
                "ssm:GetParameters"
            ],
            resources=["*"]
        )

        self.lambda_function.add_to_role_policy(custom_policy)

6.4.2 Advanced Permission Configuration

class AdvancedPermissionsStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create custom IAM role
        lambda_role = iam.Role(
            self, "LambdaExecutionRole",
            assumed_by=iam.ServicePrincipal("lambda.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name(
                    "service-role/AWSLambdaBasicExecutionRole"
                )
            ]
        )

        # Add specific permissions
        lambda_role.add_to_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=[
                    "dynamodb:GetItem",
                    "dynamodb:PutItem",
                    "dynamodb:UpdateItem",
                    "dynamodb:DeleteItem",
                    "dynamodb:Query",
                    "dynamodb:Scan"
                ],
                resources=[
                    f"arn:aws:dynamodb:{self.region}:{self.account}:table/MyTable*"
                ]
            )
        )

        # Create Lambda function using custom role
        self.secure_function = _lambda.Function(
            self, "SecureFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/secure"),
            role=lambda_role,
            function_name="secure-lambda-function"
        )

6.5 Lambda Function Versions and Aliases

6.5.1 Version Management

from aws_cdk import aws_lambda as _lambda

class VersionedLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create Lambda function
        self.main_function = _lambda.Function(
            self, "MainFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/versioned"),
            function_name="versioned-function"
        )

        # Create version
        version = self.main_function.current_version

        # Create alias pointing to version
        self.prod_alias = _lambda.Alias(
            self, "ProdAlias",
            alias_name="PROD",
            version=version
        )

        self.staging_alias = _lambda.Alias(
            self, "StagingAlias",
            alias_name="STAGING",
            version=version
        )

6.5.2 Blue-Green Deployment Configuration

from aws_cdk import aws_codedeploy as codedeploy

class BlueGreenLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create Lambda function
        self.lambda_function = _lambda.Function(
            self, "BlueGreenFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/blue_green")
        )

        # Create alias
        self.alias = _lambda.Alias(
            self, "LiveAlias",
            alias_name="live",
            version=self.lambda_function.current_version
        )

        # Create CodeDeploy application
        application = codedeploy.LambdaApplication(
            self, "CodeDeployApplication",
            application_name="my-lambda-app"
        )

        # Create deployment group
        deployment_group = codedeploy.LambdaDeploymentGroup(
            self, "DeploymentGroup",
            application=application,
            alias=self.alias,
            deployment_config=codedeploy.LambdaDeploymentConfig.LINEAR_10_PERCENT_EVERY_1_MINUTE
        )

6.6 Monitoring and Logging Configuration

6.6.1 CloudWatch Logs Configuration

from aws_cdk import (
    aws_logs as logs,
    RemovalPolicy
)

class MonitoredLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create custom log group
        log_group = logs.LogGroup(
            self, "LambdaLogGroup",
            log_group_name="/aws/lambda/monitored-function",
            retention=logs.RetentionDays.ONE_WEEK,
            removal_policy=RemovalPolicy.DESTROY
        )

        # Create Lambda function
        self.monitored_function = _lambda.Function(
            self, "MonitoredFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/monitored"),
            log_group=log_group,
            environment={
                'LOG_LEVEL': 'INFO'
            }
        )

6.6.2 X-Ray Tracing Configuration

class TrackedLambdaStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Lambda function with X-Ray tracing enabled
        self.tracked_function = _lambda.Function(
            self, "TrackedFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/tracked"),
            tracing=_lambda.Tracing.ACTIVE,
            environment={
                '_X_AMZN_TRACE_ID': 'true'
            }
        )

        # Add X-Ray permissions
        self.tracked_function.add_to_role_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=[
                    "xray:PutTraceSegments",
                    "xray:PutTelemetryRecords"
                ],
                resources=["*"]
            )
        )

6.7 Complete Example: Building a RESTful API Backend

# stacks/api_backend_stack.py
from aws_cdk import (
    Stack,
    aws_lambda as _lambda,
    aws_apigateway as apigateway,
    aws_dynamodb as dynamodb,
    Duration,
    CfnOutput
)
from constructs import Construct

class ApiBackendStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Create DynamoDB table
        self.users_table = dynamodb.Table(
            self, "UsersTable",
            table_name="users",
            partition_key=dynamodb.Attribute(
                name="userId",
                type=dynamodb.AttributeType.STRING
            ),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST
        )

        # Create Lambda function
        self.api_function = _lambda.Function(
            self, "ApiFunction",
            runtime=_lambda.Runtime.PYTHON_3_9,
            handler="index.handler",
            code=_lambda.Code.from_asset("lambda_functions/api_backend"),
            timeout=Duration.seconds(30),
            memory_size=256,
            environment={
                'USERS_TABLE': self.users_table.table_name,
                'REGION': self.region
            }
        )

        # Grant DynamoDB permissions
        self.users_table.grant_read_write_data(self.api_function)

        # Create API Gateway
        self.api = apigateway.LambdaRestApi(
            self, "UsersApi",
            handler=self.api_function,
            proxy=False,
            description="Users API powered by Lambda"
        )

        # Add resources and methods
        users_resource = self.api.root.add_resource("users")
        users_resource.add_method("GET")    # GET /users
        users_resource.add_method("POST")   # POST /users

        user_resource = users_resource.add_resource("{userId}")
        user_resource.add_method("GET")     # GET /users/{userId}
        user_resource.add_method("PUT")     # PUT /users/{userId}
        user_resource.add_method("DELETE")  # DELETE /users/{userId}

        # Output API URL
        CfnOutput(
            self, "ApiUrl",
            value=self.api.url,
            description="API Gateway URL"
        )
# lambda_functions/api_backend/index.py
import json
import boto3
import uuid
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['USERS_TABLE'])

def handler(event, context):
    """
    RESTful API handler
    """
    http_method = event['httpMethod']
    path = event['path']

    try:
        if http_method == 'GET' and path == '/users':
            return get_all_users()
        elif http_method == 'POST' and path == '/users':
            return create_user(json.loads(event['body']))
        elif http_method == 'GET' and '/users/' in path:
            user_id = path.split('/')[-1]
            return get_user(user_id)
        elif http_method == 'PUT' and '/users/' in path:
            user_id = path.split('/')[-1]
            return update_user(user_id, json.loads(event['body']))
        elif http_method == 'DELETE' and '/users/' in path:
            user_id = path.split('/')[-1]
            return delete_user(user_id)
        else:
            return {
                'statusCode': 404,
                'body': json.dumps({'error': 'Not found'})
            }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def get_all_users():
    response = table.scan()
    return {
        'statusCode': 200,
        'body': json.dumps(response['Items'])
    }

def create_user(user_data):
    user_id = str(uuid.uuid4())
    user_data['userId'] = user_id
    user_data['createdAt'] = datetime.utcnow().isoformat()

    table.put_item(Item=user_data)

    return {
        'statusCode': 201,
        'body': json.dumps(user_data)
    }

def get_user(user_id):
    response = table.get_item(Key={'userId': user_id})

    if 'Item' in response:
        return {
            'statusCode': 200,
            'body': json.dumps(response['Item'])
        }
    else:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'User not found'})
        }

def update_user(user_id, user_data):
    user_data['updatedAt'] = datetime.utcnow().isoformat()

    # Build update expression
    update_expression = "SET "
    expression_values = {}

    for key, value in user_data.items():
        if key != 'userId':
            update_expression += f"{key} = :{key}, "
            expression_values[f":{key}"] = value

    update_expression = update_expression.rstrip(', ')

    table.update_item(
        Key={'userId': user_id},
        UpdateExpression=update_expression,
        ExpressionAttributeValues=expression_values
    )

    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'User updated successfully'})
    }

def delete_user(user_id):
    table.delete_item(Key={'userId': user_id})

    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'User deleted successfully'})
    }

6.8 Deployment and Testing

6.8.1 Deployment Commands

# Install dependencies
pip install -r requirements.txt

# Synthesize CloudFormation template
cdk synth

# Deploy to AWS
cdk deploy ApiBackendStack

# View outputs
cdk deploy ApiBackendStack --outputs-file outputs.json

6.8.2 Testing API

# Get API URL
API_URL=$(cat outputs.json | jq -r '.ApiBackendStack.ApiUrl')

# Test API endpoints
curl -X GET "${API_URL}users"
curl -X POST "${API_URL}users" -H "Content-Type: application/json" -d '{"name":"John Doe","email":"john@example.com"}'

6.9 Troubleshooting and Debugging

6.9.1 Common Issues

Common Deployment Issues
  1. Permission Issues: Ensure CDK has sufficient permissions to create resources
  2. Code Packaging Issues: Check if dependencies are properly installed
  3. Timeout Issues: Adjust Lambda function timeout settings
  4. Memory Issues: Adjust memory allocation as needed

6.9.2 Debugging Tips

# Add detailed logging
import logging
import os

logger = logging.getLogger()
logger.setLevel(os.getenv('LOG_LEVEL', 'INFO'))

def handler(event, context):
    logger.info(f"Received event: {json.dumps(event)}")
    logger.info(f"Context: {context}")

    try:
        # Processing logic
        result = process_event(event)
        logger.info(f"Processing result: {result}")
        return result
    except Exception as e:
        logger.error(f"Error processing event: {str(e)}")
        raise

6.10 Chapter Summary

Key Takeaways
  • CDK provides flexible and powerful Lambda function creation and configuration capabilities
  • Proper permission configuration is the foundation for secure Lambda function operation
  • Code packaging and dependency management require special attention
  • Monitoring and logging configuration helps with production environment maintenance
  • Version and alias management supports safe deployment strategies

In the next chapter, we will explore advanced Lambda function configurations, including VPC configuration, Layer usage, error handling, and other advanced topics.

Further Reading