Aws Lambda Python Integration

AWS Lambda Python Integration

Patterns for creating high-performance AWS Lambda functions in Python with optimized cold starts and clean architecture.

Overview

AWS Lambda Python integration with two approaches: AWS Chalice (full-featured framework) and Raw Python (minimal overhead). Both support API Gateway/ALB integration with production-ready configurations.

When to Use

Use this skill when:

• Creating new Lambda functions in Python
• Migrating existing Python applications to Lambda
• Optimizing cold start performance for Python Lambda
• Choosing between framework-based and minimal Python approaches
• Configuring API Gateway or ALB integration
• Setting up deployment pipelines for Python Lambda

Instructions

1. Choose Your Approach

2. Project Structure

AWS Chalice Structure

my-chalice-app/
├── app.py                    # Main application with routes
├── requirements.txt          # Dependencies
├── .chalice/
│   ├── config.json          # Chalice configuration
│   └── deploy/              # Deployment artifacts
├── chalicelib/              # Additional modules
│   ├── __init__.py
│   └── services.py
└── tests/
    └── test_app.py

Raw Python Structure

my-lambda-function/
├── lambda_function.py       # Handler entry point
├── requirements.txt         # Dependencies
├── template.yaml            # SAM/CloudFormation template
└── src/                     # Additional modules
    ├── __init__.py
    ├── handlers.py
    └── utils.py

3. Implementation Examples

See the References section for detailed implementation guides. Quick examples:

AWS Chalice:

from chalice import Chalice
app = Chalice(app_name='my-api')

@app.route('/')
def index():
    return {'message': 'Hello from Chalice!'}

Raw Python:

def lambda_handler(event, context):
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Hello from Lambda!'})
    }

Core Concepts

Cold Start Optimization

Key strategies:

1. Initialize at module level - Persists across warm invocations 2. Use lazy loading - Defer heavy imports until needed 3. Cache boto3 clients - Reuse connections between invocations

See Raw Python Lambda for detailed patterns.

Connection Management

Create clients at module level and reuse:

_dynamodb = None

def get_table():
    global _dynamodb
    if _dynamodb is None:
        _dynamodb = boto3.resource('dynamodb').Table('my-table')
    return _dynamodb

Environment Configuration

class Config:
    TABLE_NAME = os.environ.get('TABLE_NAME')
    DEBUG = os.environ.get('DEBUG', 'false').lower() == 'true'

    @classmethod
    def validate(cls):
        if not cls.TABLE_NAME:
            raise ValueError("TABLE_NAME required")

Best Practices

Memory and Timeout Configuration

• Memory: Start with 256MB for simple handlers, 512MB for complex operations
• Timeout: Set based on expected processing time
• Simple handlers: 3-5 seconds
• API with DB calls: 10-15 seconds
• Data processing: 30-60 seconds

Dependencies

Keep requirements.txt minimal:

# Core AWS SDK - always needed
boto3>=1.35.0

# Only add what you need
requests>=2.32.0  # If calling external APIs
pydantic>=2.5.0   # If using data validation

Error Handling

Return proper HTTP codes with request ID:

def lambda_handler(event, context):
    try:
        result = process_event(event)
        return {'statusCode': 200, 'body': json.dumps(result)}
    except ValueError as e:
        return {'statusCode': 400, 'body': json.dumps({'error': str(e)})}
    except Exception as e:
        print(f"Error: {str(e)}")  # Log to CloudWatch
        return {'statusCode': 500, 'body': json.dumps({'error': 'Internal error'})}

See Raw Python Lambda for structured error patterns.

Logging

Use structured logging for CloudWatch Insights:

import logging, json
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Structured log
logger.info(json.dumps({
    'eventType': 'REQUEST',
    'requestId': context.aws_request_id,
    'path': event.get('path')
}))

See Raw Python Lambda for advanced patterns.

Deployment Options

Quick Start

Validation Checkpoint: Always run serverless print or sam validate before deploying to catch configuration errors early.

Serverless Framework:

# serverless.yml
service: my-python-api
provider:
  name: aws
  runtime: python3.12  # or python3.11
functions:
  api:
    handler: lambda_function.lambda_handler
    events:
      - http:
          path: /{proxy+}
          method: ANY

AWS SAM:

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./
      Handler: lambda_function.lambda_handler
      Runtime: python3.12  # or python3.11
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: /{proxy+}
            Method: ANY

AWS Chalice:

chalice new-project my-api
cd my-api
chalice local 8080  # Test locally before deploying
chalice deploy --stage dev

Validation Checkpoint: Test locally with chalice local or sam local invoke before deploying to production.

For complete deployment configurations including CI/CD, environment-specific settings, and advanced SAM/Serverless patterns, see Serverless Deployment.

Constraints and Warnings

Lambda Limits

• Deployment package: 250MB unzipped maximum (50MB zipped)
• Memory: 128MB to 10GB
• Timeout: 15 minutes maximum
• Concurrent executions: 1000 default (adjustable)
• Environment variables: 4KB total size

Python-Specific Considerations

• Cold start: Python has excellent cold start performance; avoid heavy imports at module level
• Dependencies: Keep requirements.txt minimal; use Lambda Layers for shared dependencies
• Native dependencies: Must be compiled for Amazon Linux 2 (x86_64 or arm64)

Common Pitfalls

1. Importing heavy libraries at module level - Defer to function level if not always needed 2. Not handling Lambda context - Use context.get_remaining_time_in_millis() for timeout awareness 3. Not validating input - Always validate and sanitize event data 4. Printing sensitive data - Be careful with logs and CloudWatch

Error Recovery: If deployment fails, check CloudWatch logs for initialization errors and run sam logs to diagnose issues.

Security Considerations

• Never hardcode credentials; use IAM roles and environment variables
• Validate all input data
• Use least privilege IAM policies
• Enable CloudTrail for audit logging

References

For detailed guidance on specific topics:

• [AWS Chalice](references/chalice-lambda.md) - Complete Chalice setup, routing, middleware, deployment
• [Raw Python Lambda](references/raw-python-lambda.md) - Minimal handler patterns, module caching, packaging
• [Serverless Deployment](references/serverless-deployment.md) - Serverless Framework, SAM, CI/CD pipelines
• [Testing Lambda](references/testing-lambda.md) - pytest, moto, SAM Local, localstack

Examples

Example 1: Create an AWS Chalice REST API

Input:

Create a Python Lambda REST API using AWS Chalice for a todo application

Process: 1. Initialize Chalice project with chalice new-project 2. Configure routes for CRUD operations 3. Set up DynamoDB integration 4. Configure deployment stages 5. Deploy with chalice deploy

Output:

• Complete Chalice project structure
• REST API with CRUD endpoints
• DynamoDB table configuration
• Deployment configuration

Example 2: Optimize Cold Start for Raw Python

Input:

My Python Lambda has slow cold start, how do I optimize it?

Process: 1. Analyze imports and initialization code 2. Move heavy imports inside functions (lazy loading) 3. Cache boto3 clients at module level 4. Remove unnecessary dependencies 5. Use provisioned concurrency if needed

Output:

• Refactored code with lazy loading
• Optimized cold start < 100ms
• Dependency analysis

Example 3: Deploy with GitHub Actions

Input:

Configure CI/CD for Python Lambda with SAM

Process: 1. Create GitHub Actions workflow 2. Set up Python environment and dependencies 3. Run pytest with coverage 4. Package with SAM 5. Deploy to dev/prod stages

Output:

• Complete .github/workflows/deploy.yml
• Multi-stage pipeline
• Integrated test automation

Version

Version: 1.0.0

AWS Chalice Lambda Framework

Complete guide for building AWS Lambda functions with the AWS Chalice framework.

What is Chalice?

AWS Chalice is a Python serverless microframework that lets you quickly create and deploy applications that use AWS Lambda. It provides:

• Decorator-based routing (similar to Flask)
• Automatic IAM policy generation
• Local development server
• Built-in CORS support
• Easy deployment to API Gateway

Installation

# Install Chalice
pip install chalice

# Verify installation
chalice --version

# Create new project
chalice new-project my-api
cd my-api

Basic Structure

Project Layout

my-chalice-project/
├── app.py                 # Main application file
├── requirements.txt       # Python dependencies
├── .chalice/
│   ├── config.json       # Stage configuration
│   └── deploy/           # Deployment artifacts (auto-generated)
├── chalicelib/           # Additional Python modules
│   ├── __init__.py
│   └── services.py
└── tests/
    └── test_app.py

Minimal Application

# app.py
from chalice import Chalice

app = Chalice(app_name='hello-world')

@app.route('/')
def index():
    return {'hello': 'world'}

Routing

HTTP Methods

from chalice import Chalice

app = Chalice(app_name='my-api')

@app.route('/users', methods=['GET'])
def list_users():
    return {'users': []}

@app.route('/users', methods=['POST'])
def create_user():
    user = app.current_request.json_body
    return {'user': user, 'created': True}

@app.route('/users/{user_id}', methods=['GET'])
def get_user(user_id):
    return {'user_id': user_id}

@app.route('/users/{user_id}', methods=['PUT'])
def update_user(user_id):
    updates = app.current_request.json_body
    return {'user_id': user_id, 'updated': updates}

@app.route('/users/{user_id}', methods=['DELETE'])
def delete_user(user_id):
    return {'user_id': user_id, 'deleted': True}

Path Parameters

@app.route('/orders/{order_id}/items/{item_id}')
def get_order_item(order_id, item_id):
    return {
        'order_id': order_id,
        'item_id': item_id
    }

Query Parameters

from urllib.parse import parse_qs

@app.route('/search')
def search():
    # Access query parameters
    query_params = app.current_request.query_params or {}

    search_term = query_params.get('q')
    page = int(query_params.get('page', 1))
    limit = int(query_params.get('limit', 10))

    return {
        'search_term': search_term,
        'page': page,
        'limit': limit
    }

Request Handling

Accessing Request Data

@app.route('/data', methods=['POST'])
def process_data():
    request = app.current_request

    # Request properties
    return {
        'method': request.method,           # POST
        'path': request.path,               # /data
        'query_params': request.query_params,
        'headers': dict(request.headers),
        'json_body': request.json_body,     # Parsed JSON body
        'raw_body': request.raw_body,       # Raw bytes
        'context': request.context,         # Lambda context
        'stage_vars': request.stage_vars,   # API Gateway stage variables
    }

Custom Responses

from chalice import Response
import json

@app.route('/custom-response')
def custom_response():
    return Response(
        body=json.dumps({'message': 'Custom response'}),
        status_code=201,
        headers={
            'Content-Type': 'application/json',
            'X-Custom-Header': 'value'
        }
    )

@app.route('/binary-data')
def binary_data():
    return Response(
        body=b'binary content',
        status_code=200,
        headers={'Content-Type': 'application/octet-stream'}
    )

CORS Configuration

Global CORS

from chalice import Chalice, CORSConfig

# Global CORS configuration
cors_config = CORSConfig(
    allow_origin='https://example.com',
    allow_headers=['Content-Type', 'Authorization'],
    allow_credentials=True,
    max_age=600
)

app = Chalice(app_name='my-api')
app.api.cors_config = cors_config

Per-Route CORS

from chalice import CORSConfig

# Simple CORS
@app.route('/public', cors=True)
def public_endpoint():
    return {'data': 'public'}

# Custom CORS per route
custom_cors = CORSConfig(
    allow_origin='https://specific-domain.com',
    allow_headers=['X-Custom-Header']
)

@app.route('/restricted', cors=custom_cors)
def restricted_endpoint():
    return {'data': 'restricted'}

Error Handling

Built-in Errors

from chalice import (
    Chalice,
    BadRequestError,
    UnauthorizedError,
    ForbiddenError,
    NotFoundError,
    ConflictError,
    UnprocessableEntityError,
    TooManyRequestsError,
    ChaliceViewError
)

app = Chalice(app_name='my-api')

@app.route('/users/{user_id}')
def get_user(user_id):
    user = find_user(user_id)
    if user is None:
        raise NotFoundError(f'User {user_id} not found')
    return user

@app.route('/users', methods=['POST'])
def create_user():
    data = app.current_request.json_body

    if not data or 'email' not in data:
        raise BadRequestError('Email is required')

    if user_exists(data['email']):
        raise ConflictError('User already exists')

    return create_new_user(data)

Custom Error Handler

from chalice import ChaliceViewError

@app.errorhandler(ChaliceViewError)
def handle_errors(error):
    return Response(
        body=json.dumps({
            'error': error.__class__.__name__,
            'message': str(error)
        }),
        status_code=error.STATUS_CODE if hasattr(error, 'STATUS_CODE') else 500,
        headers={'Content-Type': 'application/json'}
    )

AWS Service Integration

DynamoDB

import boto3
from botocore.exceptions import ClientError

# Initialize at module level for connection reuse
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('my-table')

@app.route('/items/{item_id}')
def get_item(item_id):
    try:
        response = table.get_item(Key={'id': item_id})
        item = response.get('Item')

        if not item:
            raise NotFoundError(f'Item {item_id} not found')

        return item
    except ClientError as e:
        app.log.error(f"DynamoDB error: {e}")
        raise ChaliceViewError('Database error')

@app.route('/items', methods=['POST'])
def create_item():
    item = app.current_request.json_body

    try:
        table.put_item(Item=item)
        return item
    except ClientError as e:
        app.log.error(f"DynamoDB error: {e}")
        raise ChaliceViewError('Failed to create item')

S3

import boto3
import json

s3 = boto3.client('s3')
BUCKET_NAME = 'my-bucket'

@app.route('/files/{key}')
def get_file(key):
    try:
        response = s3.get_object(Bucket=BUCKET_NAME, Key=key)
        content = response['Body'].read()
        return Response(
            body=content,
            status_code=200,
            headers={'Content-Type': response['ContentType']}
        )
    except s3.exceptions.NoSuchKey:
        raise NotFoundError(f'File {key} not found')

@app.route('/files', methods=['POST'])
def upload_file():
    request = app.current_request
    key = request.query_params.get('key')

    if not key:
        raise BadRequestError('Key parameter required')

    s3.put_object(
        Bucket=BUCKET_NAME,
        Key=key,
        Body=request.raw_body,
        ContentType=request.headers.get('content-type', 'application/octet-stream')
    )

    return {'key': key, 'uploaded': True}

SQS

import boto3
import json

sqs = boto3.client('sqs')
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789/my-queue'

@app.route('/messages', methods=['POST'])
def enqueue_message():
    message = app.current_request.json_body

    response = sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps(message),
        MessageAttributes={
            'Type': {
                'StringValue': message.get('type', 'default'),
                'DataType': 'String'
            }
        }
    )

    return {
        'message_id': response['MessageId'],
        'status': 'queued'
    }

Configuration

config.json

{
  "version": "2.0",
  "app_name": "my-api",
  "stages": {
    "dev": {
      "api_gateway_stage": "api",
      "environment_variables": {
        "DEBUG": "true",
        "TABLE_NAME": "dev-table"
      },
      "lambda_functions": {
        "api_handler": {
          "lambda_timeout": 10,
          "lambda_memory_size": 256
        }
      },
      "tags": {
        "Environment": "dev",
        "Project": "my-api"
      }
    },
    "prod": {
      "api_gateway_stage": "api",
      "environment_variables": {
        "DEBUG": "false",
        "TABLE_NAME": "prod-table"
      },
      "lambda_functions": {
        "api_handler": {
          "lambda_timeout": 30,
          "lambda_memory_size": 512,
          "reserved_concurrent_executions": 100
        }
      },
      "tags": {
        "Environment": "prod",
        "Project": "my-api"
      }
    }
  }
}

Environment Variables

import os

# Access environment variables
DEBUG = os.environ.get('DEBUG', 'false').lower() == 'true'
TABLE_NAME = os.environ.get('TABLE_NAME', 'default-table')
SECRET_KEY = os.environ.get('SECRET_KEY')

@app.route('/config')
def get_config():
    return {
        'debug': DEBUG,
        'table_name': TABLE_NAME,
        'has_secret': SECRET_KEY is not None
    }

Local Development

Local Server

# Start local development server
chalice local

# With specific port
chalice local --port 8080

# With stage configuration
chalice local --stage dev

Testing Endpoints

# Test local endpoints
curl http://localhost:8000/
curl -X POST http://localhost:8000/users \
  -H "Content-Type: application/json" \
  -d '{"name": "John", "email": "john@example.com"}'

Deployment

Deploy to AWS

# Deploy to default stage (dev)
chalice deploy

# Deploy to specific stage
chalice deploy --stage prod

# Get deployment info
chalice url --stage dev
chalice logs --stage dev

Generate CloudFormation Template

# Generate SAM/CloudFormation template
chalice package --stage prod ./out

# Output:
# ./out/sam.json          # CloudFormation template
# ./out/deployment.zip    # Lambda deployment package

Advanced Features

Request Middleware

from chalice import Chalice

app = Chalice(app_name='my-api')

@app.middleware('http')
def auth_middleware(event, get_response):
    # Run before handler
    auth_header = event.headers.get('Authorization')

    if not auth_header and event.path != '/':
        return Response(
            body=json.dumps({'error': 'Unauthorized'}),
            status_code=401
        )

    # Call the actual handler
    response = get_response(event)

    # Run after handler
    response.headers['X-Request-ID'] = event.request_context['requestId']

    return response

Scheduled Events

from chalice import Chalice, Cron

app = Chalice(app_name='scheduled-tasks')

@app.schedule(Cron(0, 12, '*', '*', '?', '*'))
def daily_report(event):
    """Run every day at 12:00 PM UTC"""
    app.log.info("Running daily report")
    generate_daily_report()
    return {'status': 'completed'}

@app.schedule('rate(1 hour)')
def hourly_cleanup(event):
    """Run every hour"""
    app.log.info("Running hourly cleanup")
    cleanup_old_data()
    return {'status': 'completed'}

S3 Event Handlers

@app.on_s3_event(bucket='my-bucket', events=['s3:ObjectCreated:*'])
def handle_s3_upload(event):
    app.log.info(f"File uploaded: {event.key}")
    process_file(event.bucket, event.key)
    return {'status': 'processed'}

SNS Event Handlers

@app.on_sns_message(topic='my-topic')
def handle_sns_message(event):
    app.log.info(f"Received SNS message: {event.subject}")
    app.log.info(f"Message body: {event.message}")
    process_notification(event.message)
    return {'status': 'processed'}

SQS Event Handlers

@app.on_sqs_message(queue='my-queue', batch_size=10)
def handle_sqs_message(event):
    for record in event:
        app.log.info(f"Processing message: {record.body}")
        process_message(record.body)
    return {'status': 'processed'}

Best Practices

1. Initialize AWS clients at module level for connection reuse across warm invocations 2. Use `chalicelib/` for additional modules to keep app.py clean 3. Configure CORS properly for browser-based clients 4. Use environment variables for configuration, not hardcoded values 5. Handle exceptions with Chalice error classes for proper HTTP responses 6. Log with `app.log` for CloudWatch integration 7. Use `chalice package` for CI/CD pipelines 8. Set appropriate timeouts and memory for your workload

Common Patterns

CRUD Service Pattern

# chalicelib/users_service.py
import boto3
from botocore.exceptions import ClientError
from chalice import NotFoundError, BadRequestError

class UsersService:
    def __init__(self, table_name):
        self.table = boto3.resource('dynamodb').Table(table_name)

    def get_user(self, user_id):
        response = self.table.get_item(Key={'id': user_id})
        user = response.get('Item')
        if not user:
            raise NotFoundError(f'User {user_id} not found')
        return user

    def create_user(self, user_data):
        if 'id' not in user_data:
            raise BadRequestError('id is required')
        self.table.put_item(Item=user_data)
        return user_data

# app.py
from chalicelib.users_service import UsersService

users_service = UsersService(os.environ.get('TABLE_NAME'))

@app.route('/users/{user_id}')
def get_user(user_id):
    return users_service.get_user(user_id)

@app.route('/users', methods=['POST'])
def create_user():
    return users_service.create_user(app.current_request.json_body)

Pagination Pattern

@app.route('/users')
def list_users():
    query_params = app.current_request.query_params or {}

    limit = int(query_params.get('limit', 20))
    cursor = query_params.get('cursor')

    kwargs = {'Limit': limit}
    if cursor:
        kwargs['ExclusiveStartKey'] = {'id': cursor}

    response = table.scan(**kwargs)

    result = {
        'users': response.get('Items', []),
        'count': response.get('Count', 0)
    }

    if 'LastEvaluatedKey' in response:
        result['next_cursor'] = response['LastEvaluatedKey']['id']

    return result

Raw Python Lambda

Guide for creating minimal AWS Lambda functions in Python without frameworks.

When to Use Raw Python

• Simple handlers with minimal dependencies
• Maximum control over the execution environment
• Smallest deployment package size
• Learning Lambda fundamentals
• Custom runtime requirements

Basic Handler

Minimal Handler

# lambda_function.py
import json

def lambda_handler(event, context):
    """
    Main entry point for Lambda function.

    Args:
        event: Event data passed to the function
        context: Lambda runtime context

    Returns:
        dict: Response object
    """
    return {
        'statusCode': 200,
        'body': json.dumps({'message': 'Hello from Lambda!'})
    }

Handler with Different Triggers

# lambda_function.py
import json

def lambda_handler(event, context):
    """Handle different event sources."""

    # Detect event source
    if 'httpMethod' in event:
        return handle_api_gateway(event, context)
    elif 'Records' in event:
        return handle_s3_event(event, context)
    elif 'source' in event and event['source'] == 'aws.events':
        return handle_cloudwatch_event(event, context)
    else:
        return handle_direct_invoke(event, context)

def handle_api_gateway(event, context):
    """Handle API Gateway proxy integration."""
    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'message': 'API Gateway request',
            'path': event.get('path'),
            'method': event.get('httpMethod')
        })
    }

def handle_s3_event(event, context):
    """Handle S3 trigger events."""
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        print(f"S3 event: {bucket}/{key}")

    return {'statusCode': 200, 'processed': len(event['Records'])}

def handle_cloudwatch_event(event, context):
    """Handle CloudWatch scheduled events."""
    print(f"Scheduled event: {event}")
    return {'statusCode': 200}

def handle_direct_invoke(event, context):
    """Handle direct Lambda invocation."""
    return {
        'statusCode': 200,
        'result': event
    }

Cold Start Optimization

Module-Level Caching

# lambda_function.py
import boto3
import os

# Initialize at module level - persists across warm invocations
_dynamodb = None
_table = None
_s3 = None

def get_dynamodb_table():
    """Lazy initialization with caching."""
    global _dynamodb, _table
    if _table is None:
        _dynamodb = boto3.resource('dynamodb')
        _table = _dynamodb.Table(os.environ['TABLE_NAME'])
    return _table

def get_s3_client():
    """Lazy initialization with caching."""
    global _s3
    if _s3 is None:
        _s3 = boto3.client('s3')
    return _s3

def lambda_handler(event, context):
    # Uses cached clients
    table = get_dynamodb_table()
    s3 = get_s3_client()

    # Handler logic here
    return {'statusCode': 200}

Lazy Loading Heavy Dependencies

# lambda_function.py
_heavy_service = None

def get_heavy_service():
    """Defer loading heavy modules until needed."""
    global _heavy_service
    if _heavy_service is None:
        # Import here to avoid loading during cold start
        from heavy_library import HeavyService
        _heavy_service = HeavyService()
    return _heavy_service

def lambda_handler(event, context):
    # Only load heavy service when needed
    if event.get('needs_heavy_processing'):
        service = get_heavy_service()
        return service.process(event['data'])

    return {'statusCode': 200, 'light': True}

API Gateway Integration

REST API Handler

# lambda_function.py
import json
import re

# Route definitions
ROUTES = {
    r'^GET /users$': 'list_users',
    r'^GET /users/(?P<user_id>[^/]+)$': 'get_user',
    r'^POST /users$': 'create_user',
    r'^PUT /users/(?P<user_id>[^/]+)$': 'update_user',
    r'^DELETE /users/(?P<user_id>[^/]+)$': 'delete_user',
}

def lambda_handler(event, context):
    """Main router for API Gateway requests."""
    http_method = event.get('httpMethod', 'GET')
    path = event.get('path', '/')

    # Match route
    route_key = f"{http_method} {path}"

    for pattern, handler_name in ROUTES.items():
        match = re.match(pattern, route_key)
        if match:
            handler = globals()[handler_name]
            return handler(event, context, **match.groupdict())

    return response(404, {'error': 'Not found'})

def list_users(event, context):
    """GET /users"""
    # Implementation here
    return response(200, {'users': []})

def get_user(event, context, user_id):
    """GET /users/{user_id}"""
    return response(200, {'user_id': user_id})

def create_user(event, context):
    """POST /users"""
    body = json.loads(event.get('body', '{}'))
    return response(201, {'created': body})

def update_user(event, context, user_id):
    """PUT /users/{user_id}"""
    body = json.loads(event.get('body', '{}'))
    return response(200, {'updated': user_id, 'data': body})

def delete_user(event, context, user_id):
    """DELETE /users/{user_id}"""
    return response(200, {'deleted': user_id})

def response(status_code, body, headers=None):
    """Helper for API Gateway responses."""
    default_headers = {
        'Content-Type': 'application/json',
        'Access-Control-Allow-Origin': '*'
    }
    if headers:
        default_headers.update(headers)

    return {
        'statusCode': status_code,
        'headers': default_headers,
        'body': json.dumps(body)
    }

Request/Response Helpers

# lambda_function.py
import json
from typing import Dict, Any, Optional

class APIGatewayRequest:
    """Wrapper for API Gateway events."""

    def __init__(self, event: Dict[str, Any]):
        self.event = event
        self.method = event.get('httpMethod', 'GET')
        self.path = event.get('path', '/')
        self.query_params = event.get('queryStringParameters') or {}
        self.path_params = event.get('pathParameters') or {}
        self.headers = {k.lower(): v for k, v in (event.get('headers') or {}).items()}

        body = event.get('body')
        self.body = json.loads(body) if body and event.get('isBase64Encoded') else body

    def get_header(self, name: str, default: str = None) -> Optional[str]:
        return self.headers.get(name.lower(), default)

    def get_query(self, name: str, default: str = None) -> Optional[str]:
        return self.query_params.get(name, default)

class APIGatewayResponse:
    """Builder for API Gateway responses."""

    def __init__(self):
        self.status_code = 200
        self.headers = {'Content-Type': 'application/json'}
        self.body = {}

    def status(self, code: int) -> 'APIGatewayResponse':
        self.status_code = code
        return self

    def header(self, name: str, value: str) -> 'APIGatewayResponse':
        self.headers[name] = value
        return self

    def json(self, data: Dict) -> Dict[str, Any]:
        return {
            'statusCode': self.status_code,
            'headers': self.headers,
            'body': json.dumps(data)
        }

    def text(self, text: str) -> Dict[str, Any]:
        self.headers['Content-Type'] = 'text/plain'
        return {
            'statusCode': self.status_code,
            'headers': self.headers,
            'body': text
        }

# Usage
def lambda_handler(event, context):
    request = APIGatewayRequest(event)

    if request.method == 'GET':
        return APIGatewayResponse().json({'path': request.path})

    return APIGatewayResponse().status(405).json({'error': 'Method not allowed'})

Error Handling

Structured Error Responses

# lambda_function.py
import json
import traceback
from typing import Dict, Any

class LambdaError(Exception):
    """Custom error with HTTP status code."""

    def __init__(self, message: str, status_code: int = 500, details: Dict = None):
        self.message = message
        self.status_code = status_code
        self.details = details or {}
        super().__init__(message)

class ValidationError(LambdaError):
    def __init__(self, message: str, details: Dict = None):
        super().__init__(message, 400, details)

class NotFoundError(LambdaError):
    def __init__(self, message: str):
        super().__init__(message, 404)

def error_response(error: LambdaError, request_id: str = None) -> Dict[str, Any]:
    """Format error for API Gateway response."""
    return {
        'statusCode': error.status_code,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({
            'error': error.__class__.__name__,
            'message': error.message,
            'details': error.details,
            'requestId': request_id
        })
    }

def lambda_handler(event, context):
    """Handler with centralized error handling."""
    try:
        return process_request(event, context)
    except LambdaError as e:
        return error_response(e, context.aws_request_id)
    except Exception as e:
        # Log full traceback for debugging
        print(f"Unhandled error: {traceback.format_exc()}")
        return error_response(
            LambdaError("Internal server error"),
            context.aws_request_id
        )

def process_request(event, context):
    """Main request processing logic."""
    user_id = event.get('pathParameters', {}).get('user_id')

    if not user_id:
        raise ValidationError("user_id is required")

    user = find_user(user_id)
    if not user:
        raise NotFoundError(f"User {user_id} not found")

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps({'user': user})
    }

def find_user(user_id: str) -> Dict:
    # Implementation
    return None

Context Usage

Lambda Context Object

# lambda_function.py

def lambda_handler(event, context):
    """Demonstrate context object usage."""
    context_info = {
        # Identity
        'function_name': context.function_name,
        'function_version': context.function_version,
        'memory_limit_mb': context.memory_limit_in_mb,
        'aws_request_id': context.aws_request_id,
        'invoked_function_arn': context.invoked_function_arn,
        'log_group_name': context.log_group_name,
        'log_stream_name': context.log_stream_name,

        # Timing
        'remaining_time_ms': context.get_remaining_time_in_millis(),
    }

    # Check if we have enough time
    if context.get_remaining_time_in_millis() < 1000:
        print("Warning: Running low on time!")

    # Identity (for Cognito authorizer)
    if hasattr(context, 'identity'):
        context_info['identity'] = {
            'cognito_identity_id': context.identity.cognito_identity_id,
            'cognito_identity_pool_id': context.identity.cognito_identity_pool_id,
        }

    # Client context (for mobile SDK)
    if hasattr(context, 'client_context'):
        context_info['client_context'] = context.client_context

    return {
        'statusCode': 200,
        'body': json.dumps(context_info)
    }

Environment Configuration

Configuration Class

# config.py
import os
from typing import List

class Config:
    """Configuration management with validation."""

    # Required environment variables
    REQUIRED = ['TABLE_NAME']

    # Optional with defaults
    DEBUG: bool = os.environ.get('DEBUG', 'false').lower() == 'true'
    REGION: str = os.environ.get('AWS_REGION', 'us-east-1')
    LOG_LEVEL: str = os.environ.get('LOG_LEVEL', 'INFO')
    TABLE_NAME: str = os.environ.get('TABLE_NAME', '')
    BUCKET_NAME: str = os.environ.get('BUCKET_NAME', '')

    # Numeric values
    TIMEOUT_SECONDS: int = int(os.environ.get('TIMEOUT_SECONDS', '30'))
    MAX_RETRIES: int = int(os.environ.get('MAX_RETRIES', '3'))

    # Lists
    ALLOWED_ORIGINS: List[str] = os.environ.get('ALLOWED_ORIGINS', '*').split(',')

    @classmethod
    def validate(cls) -> None:
        """Validate required configuration."""
        missing = []
        for var in cls.REQUIRED:
            if not getattr(cls, var):
                missing.append(var)

        if missing:
            raise ValueError(f"Missing required environment variables: {missing}")

    @classmethod
    def to_dict(cls) -> dict:
        """Export configuration as dictionary."""
        return {
            'DEBUG': cls.DEBUG,
            'REGION': cls.REGION,
            'LOG_LEVEL': cls.LOG_LEVEL,
            'TABLE_NAME': cls.TABLE_NAME,
            'BUCKET_NAME': cls.BUCKET_NAME,
            'TIMEOUT_SECONDS': cls.TIMEOUT_SECONDS,
            'MAX_RETRIES': cls.MAX_RETRIES,
        }

# lambda_function.py
from config import Config

# Validate on module load
Config.validate()

def lambda_handler(event, context):
    if Config.DEBUG:
        print(f"Config: {Config.to_dict()}")

    return {'statusCode': 200, 'config': Config.to_dict()}

Logging

Structured Logging

# lambda_function.py
import json
import logging
from datetime import datetime, timezone

# Configure logger
logger = logging.getLogger()
logger.setLevel(logging.INFO)

class StructuredLog:
    """Structured logging for CloudWatch."""

    @staticmethod
    def info(message: str, **kwargs):
        log_entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'level': 'INFO',
            'message': message,
            **kwargs
        }
        logger.info(json.dumps(log_entry))

    @staticmethod
    def error(message: str, error: Exception = None, **kwargs):
        log_entry = {
            'timestamp': datetime.now(timezone.utc).isoformat(),
            'level': 'ERROR',
            'message': message,
            **kwargs
        }
        if error:
            log_entry['error_type'] = error.__class__.__name__
            log_entry['error_message'] = str(error)

        logger.error(json.dumps(log_entry))

    @staticmethod
    def metric(name: str, value: float, unit: str = 'Count'):
        """Emit CloudWatch metric via log."""
        log_entry = {
            '_aws': {
                'Timestamp': int(datetime.now(timezone.utc).timestamp() * 1000),
                'CloudWatchMetrics': [{
                    'Namespace': 'Lambda/Custom',
                    'Dimensions': [['FunctionName']],
                    'Metrics': [{'Name': name, 'Unit': unit}]
                }]
            },
            'FunctionName': os.environ.get('AWS_LAMBDA_FUNCTION_NAME', 'unknown'),
            name: value
        }
        logger.info(json.dumps(log_entry))

# Usage
def lambda_handler(event, context):
    StructuredLog.info(
        'Processing request',
        request_id=context.aws_request_id,
        path=event.get('path')
    )

    try:
        result = process_event(event)
        StructuredLog.metric('SuccessCount', 1)
        return {'statusCode': 200, 'body': result}
    except Exception as e:
        StructuredLog.error('Processing failed', error=e)
        StructuredLog.metric('ErrorCount', 1)
        raise

Deployment Packaging

requirements.txt

# Minimal requirements for raw Python Lambda
boto3>=1.35.0

# Add only what you need
requests>=2.32.0  # For HTTP calls
pydantic>=2.5.0   # For data validation

Build Script

#!/bin/bash
# build.sh - Build deployment package

set -e

PACKAGE_DIR="package"
OUTPUT="deployment.zip"

# Clean previous builds
rm -rf $PACKAGE_DIR $OUTPUT

# Install dependencies
pip install -r requirements.txt -t $PACKAGE_DIR

# Copy source code
cp lambda_function.py config.py $PACKAGE_DIR/

# Create zip
cd $PACKAGE_DIR
zip -r ../$OUTPUT .
cd ..

echo "Created $OUTPUT"

Best Practices

1. Keep handlers small - Delegate to separate functions/classes 2. Initialize outside handler - Module-level for warm start reuse 3. Use lazy loading - Defer heavy imports until needed 4. Handle timeouts - Check context.get_remaining_time_in_millis() 5. Validate input - Always check event structure 6. Use structured logging - JSON for CloudWatch Insights 7. Environment configuration - No hardcoded values 8. Error handling - Graceful degradation with proper HTTP codes

Serverless Deployment for Python Lambda

Deployment patterns for Python Lambda functions using Serverless Framework, AWS SAM, and CI/CD pipelines.

Serverless Framework

Basic Configuration

# serverless.yml
service: my-python-api

provider:
  name: aws
  runtime: python3.12  # or python3.11
  stage: ${opt:stage, 'dev'}
  region: ${opt:region, 'us-east-1'}
  memorySize: 256
  timeout: 10

  # Environment variables for all functions
  environment:
    STAGE: ${self:provider.stage}
    REGION: ${self:provider.region}
    TABLE_NAME: ${self:service}-table-${self:provider.stage}

  # IAM permissions
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - logs:CreateLogGroup
            - logs:CreateLogStream
            - logs:PutLogEvents
          Resource: '*'

functions:
  api:
    handler: lambda_function.lambda_handler
    events:
      - http:
          path: /{proxy+}
          method: ANY
          cors: true

Advanced Configuration

# serverless.yml
service: my-python-api

provider:
  name: aws
  runtime: python3.12  # or python3.11
  stage: ${opt:stage, 'dev'}
  region: ${opt:region, 'us-east-1'}
  memorySize: 256
  timeout: 10
  logRetentionInDays: 14
  versionFunctions: false

  # VPC configuration
  vpc:
    securityGroupIds:
      - sg-xxxxxxxx
    subnetIds:
      - subnet-xxxxxxxx
      - subnet-yyyyyyyy

  # Environment variables
  environment:
    STAGE: ${self:provider.stage}
    REGION: ${self:provider.region}
    TABLE_NAME: !Ref UsersTable
    BUCKET_NAME: !Ref AssetsBucket

  # IAM role statements
  iam:
    role:
      name: ${self:service}-${self:provider.stage}-role
      statements:
        # DynamoDB permissions
        - Effect: Allow
          Action:
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
            - dynamodb:DeleteItem
            - dynamodb:Query
            - dynamodb:Scan
          Resource:
            - !GetAtt UsersTable.Arn
            - !Sub "${UsersTable.Arn}/index/*"

        # S3 permissions
        - Effect: Allow
          Action:
            - s3:GetObject
            - s3:PutObject
            - s3:DeleteObject
          Resource:
            - !Sub "${AssetsBucket.Arn}/*"

        # CloudWatch permissions
        - Effect: Allow
          Action:
            - cloudwatch:PutMetricData
          Resource: '*'

  # API Gateway settings
  apiGateway:
    binaryMediaTypes:
      - 'multipart/form-data'
    minimumCompressionSize: 1024

plugins:
  - serverless-python-requirements
  - serverless-offline

custom:
  pythonRequirements:
    dockerizePip: true
    slim: true
    strip: false
    layer: false

  serverless-offline:
    httpPort: 3000
    lambdaPort: 3002

package:
  individually: false
  patterns:
    - '!.git/**'
    - '!.gitignore'
    - '!.DS_Store'
    - '!node_modules/**'
    - '!.pytest_cache/**'
    - '!tests/**'
    - '!.env'
    - '!.venv/**'
    - '!venv/**'
    - '!__pycache__/**'
    - '!.mypy_cache/**'

functions:
  api:
    handler: lambda_function.lambda_handler
    description: Main API handler
    memorySize: 512
    timeout: 30
    reservedConcurrency: 100
    provisionedConcurrency: 10
    environment:
      FUNCTION_NAME: api
    events:
      - http:
          path: /{proxy+}
          method: ANY
          cors:
            origin: '*'
            headers:
              - Content-Type
              - Authorization
              - X-Amz-Date
              - X-Api-Key
              - X-Amz-Security-Token
            allowCredentials: true

  scheduled-task:
    handler: handlers.scheduled_task
    description: Daily scheduled task
    events:
      - schedule: rate(1 day)

  sqs-processor:
    handler: handlers.process_sqs_message
    description: SQS message processor
    reservedConcurrency: 50
    events:
      - sqs:
          arn: !GetAtt MessageQueue.Arn
          batchSize: 10
          maximumBatchingWindowInSeconds: 5

resources:
  Resources:
    UsersTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:service}-users-${self:provider.stage}
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
          - AttributeName: email
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        GlobalSecondaryIndexes:
          - IndexName: email-index
            KeySchema:
              - AttributeName: email
                KeyType: HASH
            Projection:
              ProjectionType: ALL

    AssetsBucket:
      Type: AWS::S3::Bucket
      Properties:
        BucketName: ${self:service}-assets-${self:provider.stage}
        PublicAccessBlockConfiguration:
          BlockPublicAcls: true
          BlockPublicPolicy: true
          IgnorePublicAcls: true
          RestrictPublicBuckets: true

    MessageQueue:
      Type: AWS::SQS::Queue
      Properties:
        QueueName: ${self:service}-messages-${self:provider.stage}
        VisibilityTimeout: 300

  Outputs:
    ApiGatewayRestApiId:
      Value: !Ref ApiGatewayRestApi
      Export:
        Name: ${self:service}-${self:provider.stage}-restApiId

    ApiGatewayRestApiRootResourceId:
      Value: !GetAtt ApiGatewayRestApi.RootResourceId
      Export:
        Name: ${self:service}-${self:provider.stage}-rootResourceId

Commands

# Install plugins
npm install

# Deploy to dev
serverless deploy

# Deploy to specific stage
serverless deploy --stage prod

# Deploy specific function
serverless deploy function -f api

# Invoke locally
serverless invoke local -f api -p event.json

# View logs
serverless logs -f api --tail

# Remove stack
serverless remove

AWS SAM

Basic Template

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Python Lambda API

Globals:
  Function:
    Timeout: 10
    MemorySize: 256
    Runtime: python3.12  # or python3.11
    Architectures:
      - x86_64
    Environment:
      Variables:
        LOG_LEVEL: INFO

Parameters:
  Stage:
    Type: String
    Default: dev
    AllowedValues:
      - dev
      - staging
      - prod

Conditions:
  IsProd: !Equals [!Ref Stage, prod]

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./
      Handler: lambda_function.lambda_handler
      Description: Main API handler
      MemorySize: 512
      Timeout: 30
      AutoPublishAlias: live
      ProvisionedConcurrencyConfig:
        !If [IsProd, {ProvisionedConcurrentExecutions: 10}, !Ref "AWS::NoValue"]
      Environment:
        Variables:
          TABLE_NAME: !Ref UsersTable
          STAGE: !Ref Stage
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: /{proxy+}
            Method: ANY
            RestApiId: !Ref ApiGatewayApi
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UsersTable

  ApiGatewayApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: !Ref Stage
      Cors:
        AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization'"
        AllowOrigin: "'*'"
        MaxAge: "'600'"

  UsersTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: !Sub "${AWS::StackName}-users"
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH

Outputs:
  ApiUrl:
    Description: API Gateway endpoint URL
    Value: !Sub "https://${ApiGatewayApi}.execute-api.${AWS::Region}.amazonaws.com/${Stage}/"

  ApiFunctionArn:
    Description: Lambda function ARN
    Value: !GetAtt ApiFunction.Arn

SAM with Lambda Layers

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Timeout: 10
    Runtime: python3.12  # or python3.11
    Layers:
      - !Ref DependenciesLayer

Resources:
  DependenciesLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      LayerName: python-dependencies
      Description: Common Python dependencies
      ContentUri: dependencies/
      CompatibleRuntimes:
        - python3.12
        - python3.10
      RetentionPolicy: Retain
    Metadata:
      BuildMethod: python3.12

  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.lambda_handler
      Environment:
        Variables:
          TABLE_NAME: !Ref UsersTable
      Events:
        ApiEvent:
          Type: Api
          Properties:
            Path: /{proxy+}
            Method: ANY

SAM Commands

# Build the application
sam build

# Build with specific runtime
sam build --use-container

# Local invoke
sam local invoke ApiFunction -e events/api.json

# Local API server
sam local start-api

# Validate template
sam validate

# Deploy (guided)
sam deploy --guided

# Deploy (CI/CD)
sam deploy \
  --stack-name my-stack \
  --s3-bucket my-deployment-bucket \
  --region us-east-1 \
  --capabilities CAPABILITY_IAM \
  --parameter-overrides Stage=prod

# Delete stack
sam delete

AWS Chalice Deployment

Chalice Config

{
  "version": "2.0",
  "app_name": "my-api",
  "stages": {
    "dev": {
      "api_gateway_stage": "api",
      "environment_variables": {
        "DEBUG": "true"
      },
      "lambda_functions": {
        "api_handler": {
          "lambda_timeout": 10,
          "lambda_memory_size": 256,
          "tags": {
            "Environment": "dev"
          }
        }
      }
    },
    "prod": {
      "api_gateway_stage": "api",
      "environment_variables": {
        "DEBUG": "false"
      },
      "lambda_functions": {
        "api_handler": {
          "lambda_timeout": 30,
          "lambda_memory_size": 512,
          "reserved_concurrent_executions": 100,
          "provisioned_concurrency": 10
        }
      },
      "tags": {
        "Environment": "prod"
      }
    }
  }
}

Chalice Deployment Commands

# Deploy to dev
chalice deploy

# Deploy to prod
chalice deploy --stage prod

# Generate CloudFormation template
chalice package --stage prod ./out

# Deploy with CloudFormation
cd out
aws cloudformation deploy \
  --template-file sam.json \
  --stack-name my-api-prod \
  --capabilities CAPABILITY_IAM

CI/CD Pipelines

GitHub Actions - SAM

# .github/workflows/deploy.yml
name: Deploy Lambda

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  AWS_REGION: us-east-1
  PYTHON_VERSION: '3.11'

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Cache pip dependencies
        uses: actions/cache@v4
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-dev.txt

      - name: Run linting
        run: |
          flake8 .
          black --check .
          isort --check-only .

      - name: Run type checking
        run: mypy .

      - name: Run tests
        run: pytest --cov=. --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v4
        with:
          files: ./coverage.xml
          token: ${{ secrets.CODECOV_TOKEN }}

  deploy-dev:
    needs: test
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment: dev
    steps:
      - uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ env.AWS_REGION }}

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install SAM CLI
        run: |
          pip install aws-sam-cli

      - name: Build
        run: sam build --use-container

      - name: Deploy to Dev
        run: |
          sam deploy \
            --stack-name my-api-dev \
            --s3-bucket ${{ secrets.DEPLOYMENT_BUCKET }} \
            --region ${{ env.AWS_REGION }} \
            --capabilities CAPABILITY_IAM \
            --parameter-overrides Stage=dev \
            --no-confirm-changeset \
            --no-fail-on-empty-changeset

  deploy-prod:
    needs: deploy-dev
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    environment: prod
    steps:
      - uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ env.AWS_REGION }}

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install SAM CLI
        run: |
          pip install aws-sam-cli

      - name: Build
        run: sam build --use-container

      - name: Deploy to Prod
        run: |
          sam deploy \
            --stack-name my-api-prod \
            --s3-bucket ${{ secrets.DEPLOYMENT_BUCKET }} \
            --region ${{ env.AWS_REGION }} \
            --capabilities CAPABILITY_IAM \
            --parameter-overrides Stage=prod \
            --no-confirm-changeset \
            --no-fail-on-empty-changeset

GitHub Actions - Serverless Framework

# .github/workflows/deploy-serverless.yml
name: Deploy with Serverless

on:
  push:
    branches: [main]

env:
  NODE_VERSION: '18'
  PYTHON_VERSION: '3.11'

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest

      - name: Run tests
        run: pytest

  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Node.js
        uses: actions/setup-node@v4
        with:
          node-version: ${{ env.NODE_VERSION }}

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}

      - name: Install Serverless
        run: npm install -g serverless

      - name: Install plugins
        run: npm install

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1

      - name: Deploy to Dev
        if: github.ref != 'refs/heads/main'
        run: serverless deploy --stage dev

      - name: Deploy to Prod
        if: github.ref == 'refs/heads/main'
        run: serverless deploy --stage prod

Deployment Best Practices

Stages and Environments

# serverless.yml
provider:
  stage: ${opt:stage, 'dev'}

  environment:
    STAGE: ${self:provider.stage}
    LOG_LEVEL: ${self:custom.logLevel.${self:provider.stage}}

custom:
  logLevel:
    dev: DEBUG
    staging: INFO
    prod: WARN

Secrets Management

# serverless.yml
provider:
  environment:
    DATABASE_URL: ${ssm:/${self:service}/${self:provider.stage}/database-url}
    API_KEY: ${ssm:/${self:service}/${self:provider.stage}/api-key~true}  # SecureString

Rollback Strategy

# serverless.yml
provider:
  versionFunctions: true
  deploymentSettings:
    alias: live
    type: AllAtOnce
    # For canary deployment:
    # type: Canary10Percent5Minutes

Monitoring and Alarms

# serverless.yml
resources:
  Resources:
    ErrorAlarm:
      Type: AWS::CloudWatch::Alarm
      Properties:
        AlarmName: ${self:service}-${self:provider.stage}-errors
        MetricName: Errors
        Namespace: AWS/Lambda
        Statistic: Sum
        Period: 60
        EvaluationPeriods: 1
        Threshold: 1
        ComparisonOperator: GreaterThanOrEqualToThreshold
        Dimensions:
          - Name: FunctionName
            Value: ${self:service}-${self:provider.stage}-api

Comparison

Testing Python Lambda Functions

Testing strategies for Python Lambda functions including unit tests, integration tests, and local emulation.

Testing Tools

# requirements-dev.txt
pytest>=8.0.0
pytest-cov>=5.0.0
pytest-asyncio>=0.23.0
moto>=5.0.0  # AWS service mocking - uses mock_aws decorator
responses>=0.25.0  # HTTP mocking
factory-boy>=3.3.0  # Test data generation
freezegun>=1.5.0  # Time mocking

Unit Testing

Basic Test Structure

# tests/test_lambda_function.py
import json
import pytest
from unittest.mock import Mock, patch, MagicMock

from lambda_function import lambda_handler, get_user, create_user

class TestLambdaHandler:
    """Test suite for Lambda handler."""

    @pytest.fixture
    def lambda_context(self):
        """Create a mock Lambda context."""
        context = Mock()
        context.function_name = 'test-function'
        context.memory_limit_in_mb = 256
        context.invoked_function_arn = 'arn:aws:lambda:us-east-1:123456789:function:test-function'
        context.aws_request_id = 'test-request-id'
        context.get_remaining_time_in_millis.return_value = 5000
        return context

    @pytest.fixture
    def api_gateway_event(self):
        """Create a sample API Gateway event."""
        return {
            'httpMethod': 'GET',
            'path': '/users/123',
            'pathParameters': {'user_id': '123'},
            'queryStringParameters': None,
            'headers': {'Content-Type': 'application/json'},
            'body': None,
            'requestContext': {
                'requestId': 'test-request-id',
                'identity': {'sourceIp': '127.0.0.1'}
            }
        }

    def test_lambda_handler_get_user(self, api_gateway_event, lambda_context):
        """Test GET /users/{id} endpoint."""
        with patch('lambda_function.get_table') as mock_get_table:
            mock_table = Mock()
            mock_table.get_item.return_value = {'Item': {'id': '123', 'name': 'Test User'}}
            mock_get_table.return_value = mock_table

            response = lambda_handler(api_gateway_event, lambda_context)

            assert response['statusCode'] == 200
            body = json.loads(response['body'])
            assert body['id'] == '123'
            assert body['name'] == 'Test User'

    def test_lambda_handler_user_not_found(self, api_gateway_event, lambda_context):
        """Test 404 response when user not found."""
        with patch('lambda_function.get_table') as mock_get_table:
            mock_table = Mock()
            mock_table.get_item.return_value = {}
            mock_get_table.return_value = mock_table

            response = lambda_handler(api_gateway_event, lambda_context)

            assert response['statusCode'] == 404
            body = json.loads(response['body'])
            assert 'error' in body

Testing with Fixtures

# tests/conftest.py
import pytest
import json
from unittest.mock import Mock

@pytest.fixture
def mock_context():
    """Standard mock Lambda context."""
    context = Mock()
    context.function_name = 'test-function'
    context.memory_limit_in_mb = 256
    context.invoked_function_arn = 'arn:aws:lambda:us-east-1:123456789:function:test'
    context.aws_request_id = 'test-request-id'
    context.get_remaining_time_in_millis.return_value = 30000
    return context

@pytest.fixture
def mock_api_event():
    """Factory for API Gateway events."""
    def _make_event(
        method='GET',
        path='/',
        path_params=None,
        query_params=None,
        body=None,
        headers=None
    ):
        event = {
            'httpMethod': method,
            'path': path,
            'pathParameters': path_params or {},
            'queryStringParameters': query_params or {},
            'headers': headers or {'Content-Type': 'application/json'},
            'body': json.dumps(body) if body else None,
            'requestContext': {'requestId': 'test-id'}
        }
        return event
    return _make_event

@pytest.fixture
def mock_dynamodb_item():
    """Factory for DynamoDB items."""
    def _make_item(user_id='123', name='Test', email='test@example.com'):
        return {
            'id': user_id,
            'name': name,
            'email': email,
            'created_at': '2024-01-01T00:00:00Z'
        }
    return _make_item

Moto: AWS Service Mocking

DynamoDB Testing

# tests/test_with_moto.py
import pytest
import boto3
from moto import mock_aws
import json

from lambda_function import lambda_handler

@pytest.fixture
def dynamodb_table():
    """Create mock DynamoDB table."""
    with mock_aws():
        dynamodb = boto3.resource('dynamodb', region_name='us-east-1')

        table = dynamodb.create_table(
            TableName='test-users',
            KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
            AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
            BillingMode='PAY_PER_REQUEST'
        )
        table.wait_until_exists()

        # Seed with test data
        table.put_item(Item={'id': '123', 'name': 'Test User', 'email': 'test@example.com'})

        yield table

def test_get_user_with_moto(dynamodb_table, mock_context):
    """Test using mocked DynamoDB."""
    from lambda_function import get_table

    # Override the table in lambda_function
    import lambda_function
    lambda_function._table = dynamodb_table

    event = {
        'httpMethod': 'GET',
        'path': '/users/123',
        'pathParameters': {'user_id': '123'}
    }

    response = lambda_handler(event, mock_context)

    assert response['statusCode'] == 200
    body = json.loads(response['body'])
    assert body['name'] == 'Test User'

@mock_aws
def test_create_user():
    """Test using decorator style."""
    # Setup
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.create_table(
        TableName='test-users',
        KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )

    # Test
    from lambda_function import create_user
    import lambda_function
    lambda_function._table = table

    result = create_user({'id': '456', 'name': 'New User'})

    # Verify
    response = table.get_item(Key={'id': '456'})
    assert response['Item']['name'] == 'New User'

S3 Testing

# tests/test_s3_operations.py
import pytest
import boto3
from moto import mock_aws
import json

@mock_aws
def test_s3_upload_and_retrieve():
    """Test S3 operations."""
    # Setup
    s3 = boto3.client('s3', region_name='us-east-1')
    bucket_name = 'test-bucket'
    s3.create_bucket(Bucket=bucket_name)

    # Test upload
    s3.put_object(
        Bucket=bucket_name,
        Key='test-file.json',
        Body=json.dumps({'key': 'value'})
    )

    # Test retrieve
    response = s3.get_object(Bucket=bucket_name, Key='test-file.json')
    content = json.loads(response['Body'].read())

    assert content['key'] == 'value'

@mock_aws
def test_lambda_s3_trigger():
    """Test Lambda triggered by S3 event."""
    from lambda_function import lambda_handler

    s3_event = {
        'Records': [{
            'eventVersion': '2.1',
            'eventSource': 'aws:s3',
            'awsRegion': 'us-east-1',
            'eventName': 'ObjectCreated:Put',
            's3': {
                'bucket': {'name': 'test-bucket'},
                'object': {'key': 'uploads/file.txt'}
            }
        }]
    }

    context = Mock()
    response = lambda_handler(s3_event, context)

    assert response['statusCode'] == 200

SQS Testing

# tests/test_sqs.py
import pytest
import boto3
from moto import mock_aws
import json

@mock_aws
def test_sqs_send_and_receive():
    """Test SQS operations."""
    sqs = boto3.client('sqs', region_name='us-east-1')

    # Create queue
    queue = sqs.create_queue(QueueName='test-queue')
    queue_url = queue['QueueUrl']

    # Send message
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({'task': 'process_data'})
    )

    # Receive message
    messages = sqs.receive_message(QueueUrl=queue_url)
    body = json.loads(messages['Messages'][0]['Body'])

    assert body['task'] == 'process_data'

Testing AWS Chalice

Chalice Test Client

# tests/test_chalice_app.py
import pytest
import json
from chalice.test import Client
from moto import mock_aws
import boto3

from app import app

@pytest.fixture
def client():
    """Create Chalice test client."""
    with Client(app) as client:
        yield client

@mock_aws
def test_index_endpoint(client):
    """Test GET / endpoint."""
    response = client.http.get('/')

    assert response.status_code == 200
    assert response.json_body == {'message': 'Hello from Chalice!'}

@mock_aws
def test_create_user(client):
    """Test POST /users endpoint."""
    # Setup mock DynamoDB
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.create_table(
        TableName='users',
        KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    table.wait_until_exists()

    response = client.http.post(
        '/users',
        body=json.dumps({'id': '123', 'name': 'Test User'}),
        headers={'Content-Type': 'application/json'}
    )

    assert response.status_code == 201
    assert 'id' in response.json_body

def test_404_response(client):
    """Test 404 handling."""
    response = client.http.get('/nonexistent')

    assert response.status_code == 404

Testing Chalice Events

# tests/test_chalice_events.py
import pytest
from chalice.test import Client
from app import app

@pytest.fixture
def client():
    with Client(app) as client:
        yield client

def test_s3_event_handler(client):
    """Test S3 event handler."""
    event = client.events.generate_s3_event(
        bucket='my-bucket',
        key='uploads/file.txt'
    )

    response = client.lambda_.invoke(
        'handle_s3_upload',
        event
    )

    assert response.payload == {'status': 'processed'}

def test_scheduled_event(client):
    """Test CloudWatch scheduled event."""
    event = client.events.generate_cw_event(
        source='aws.events',
        detail_type='Scheduled Event'
    )

    response = client.lambda_.invoke('daily_report', event)

    assert response.payload['status'] == 'completed'

def test_sns_event(client):
    """Test SNS event handler."""
    event = client.events.generate_sns_event(
        message=json.dumps({'notification': 'test'}),
        subject='Test Subject'
    )

    response = client.lambda_.invoke('handle_sns_message', event)

    assert response.payload['status'] == 'processed'

Local Testing

SAM Local

# Install SAM CLI
pip install aws-sam-cli

# Build the application
sam build

# Start local API
sam local start-api

# Invoke function locally
sam local invoke ApiFunction -e events/api-get.json

# Invoke with environment variables
sam local invoke ApiFunction \
  -e events/api-get.json \
  --env-vars env.json

Serverless Offline

# Install plugin
npm install serverless-offline

# Start offline server
serverless offline

# With specific port
serverless offline --httpPort 3000

# Invoke function
serverless invoke local -f api -p event.json

LocalStack

# docker-compose.yml
version: '3.8'
services:
  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=lambda,s3,dynamodb,apigateway,sqs
      - DEBUG=1
      - LAMBDA_EXECUTOR=docker
    volumes:
      - "/var/run/docker.sock:/var/run/docker.sock"

# tests/test_with_localstack.py
import pytest
import boto3
import requests
import json

LOCALSTACK_ENDPOINT = 'http://localhost:4566'

@pytest.fixture(scope='module')
def aws_clients():
    """Create AWS clients pointing to LocalStack."""
    return {
        'lambda': boto3.client(
            'lambda',
            endpoint_url=LOCALSTACK_ENDPOINT,
            region_name='us-east-1',
            aws_access_key_id='test',
            aws_secret_access_key='test'
        ),
        'dynamodb': boto3.resource(
            'dynamodb',
            endpoint_url=LOCALSTACK_ENDPOINT,
            region_name='us-east-1',
            aws_access_key_id='test',
            aws_secret_access_key='test'
        ),
        'apigateway': boto3.client(
            'apigateway',
            endpoint_url=LOCALSTACK_ENDPOINT,
            region_name='us-east-1',
            aws_access_key_id='test',
            aws_secret_access_key='test'
        )
    }

def test_deploy_and_invoke(aws_clients):
    """Deploy Lambda and invoke via API Gateway."""
    # This would deploy the Lambda and test end-to-end
    pass

Integration Testing

Full Stack Test

# tests/integration/test_api_integration.py
import pytest
import requests
import os

# Skip if no deployment
pytestmark = pytest.mark.skipif(
    not os.getenv('API_ENDPOINT'),
    reason='API_ENDPOINT not set'
)

@pytest.fixture
def api_endpoint():
    return os.getenv('API_ENDPOINT')

@pytest.fixture
def api_key():
    return os.getenv('API_KEY')

def test_health_check(api_endpoint):
    """Test health endpoint."""
    response = requests.get(f"{api_endpoint}/health")
    assert response.status_code == 200
    assert response.json()['status'] == 'ok'

def test_create_and_get_user(api_endpoint, api_key):
    """Test full CRUD flow."""
    headers = {'X-Api-Key': api_key}

    # Create user
    create_response = requests.post(
        f"{api_endpoint}/users",
        json={'name': 'Integration Test', 'email': 'test@example.com'},
        headers=headers
    )
    assert create_response.status_code == 201
    user_id = create_response.json()['id']

    # Get user
    get_response = requests.get(
        f"{api_endpoint}/users/{user_id}",
        headers=headers
    )
    assert get_response.status_code == 200
    assert get_response.json()['name'] == 'Integration Test'

    # Cleanup
    requests.delete(f"{api_endpoint}/users/{user_id}", headers=headers)

Code Coverage

# Run tests with coverage
pytest --cov=lambda_function --cov-report=term-missing

# Generate HTML report
pytest --cov=lambda_function --cov-report=html

# Generate XML report for CI
pytest --cov=lambda_function --cov-report=xml

# Fail if coverage below threshold
pytest --cov=lambda_function --cov-fail-under=80

Coverage Configuration

# .coveragerc
[run]
source = .
omit =
    */tests/*
    */venv/*
    */.venv/*
    */__pycache__/*

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise NotImplementedError
    if __name__ == .__main__.:
    pass

Testing Best Practices

1. Use fixtures - For common setup and teardown 2. Mock external services - Use moto for AWS services 3. Test error paths - Not just happy paths 4. Use parameterized tests - For multiple test cases 5. Keep tests fast - Unit tests should run in milliseconds 6. Test in isolation - Each test should be independent 7. Use factories - For test data generation 8. Clean up resources - Even in tests