Skip to main content

Chapter 6: Creating Lambda Functions with CDK

Haiyue9/1/25About 7 minaws

Chapter 6: Creating Lambda Functions with CDK

Chapter Overview

This chapter will provide a detailed introduction to creating, configuring, and deploying Lambda functions using the Python CDK. We will start with the simplest Hello World function and gradually delve into complex function configurations and deployment strategies.

Learning Objectives

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

6.1 Lambda Function Basic Configuration

6.1.1 The 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 the 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 a Lambda Function from a 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 a Lambda function from a 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 functions with multiple versions
        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 permission to access the 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:
        # Use the requests library
        response = requests.get('https://httpbin.org/json', timeout=10)
        external_data = response.json()
        
        # Process data with pandas
        df = pd.DataFrame([
            {'name': 'Alice', 'age': 25},
            {'name': 'Bob', 'age': 30},
            {'name': 'Charlie', 'age': 35}
        ])
        
        # Use 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 Building with Docker

# 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 the function code
COPY index.py ${LAMBDA_TASK_ROOT}

# Set the 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 a Lambda function using a 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 an S3 bucket
        self.bucket = s3.Bucket(
            self, "MyBucket",
            bucket_name="my-lambda-cdk-bucket"
        )
        
        # Create a DynamoDB table
        self.table = dynamodb.Table(
            self, "MyTable",
            table_name="my-lambda-table",
            partition_key=dynamodb.Attribute(
                name="id",
                type=dynamodb.AttributeType.STRING
            )
        )
        
        # Create a 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 a 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 a 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 a Lambda function with the 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 a 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 a version
        version = self.main_function.current_version
        
        # Create an alias pointing to the 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 a 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 an alias
        self.alias = _lambda.Alias(
            self, "LiveAlias",
            alias_name="live",
            version=self.lambda_function.current_version
        )
        
        # Create a CodeDeploy application
        application = codedeploy.LambdaApplication(
            self, "CodeDeployApplication",
            application_name="my-lambda-app"
        )
        
        # Create a 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 a 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 a 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 a 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 a 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 an 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 the 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 the 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 the CloudFormation template
cdk synth

# Deploy to AWS
cdk deploy ApiBackendStack

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

6.8.2 Testing the API

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

# Test the 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 that dependencies are installed correctly
  3. Timeout Issues: Adjust the Lambda function's timeout setting
  4. Memory Issues: Adjust the 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:
        # Process 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 a powerful and flexible way to create and configure Lambda functions
  • Proper permission configuration is fundamental to the secure operation of Lambda functions
  • Special attention should be paid to code packaging and dependency management
  • Monitoring and logging configurations are helpful for maintaining production environments
  • Version and alias management support safe deployment strategies

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

Further Reading

Last Updated
Contributors: Haiyue
Default footer
Copyright © 2025 Haiyue