Part 3: Building Your First Lambda Function

From Learning to Doing

After understanding the theory, I remember being eager to build my first Lambda function. This guide walks you through creating a practical, production-ready function based on my actual projects.

What We'll Build

We're creating a User Registration API that:

• Validates user input

• Stores data in DynamoDB

• Sends confirmation emails via SNS

• Returns appropriate HTTP responses

This mirrors a real feature I built for a SaaS application.

Prerequisites Setup

Before coding, ensure you have these tools installed. I use these daily:

Install AWS CLI

# macOS
brew install awscli

# Verify installation
aws --version

Configure AWS Credentials

aws configure
# Enter:
# - AWS Access Key ID
# - AWS Secret Access Key
# - Default region (e.g., us-east-1)
# - Output format: json

Install AWS SAM CLI

# macOS
brew install aws-sam-cli

# Verify installation
sam --version

Project Structure

Here's the directory structure I use for all my Lambda projects:

user-registration-api/
├── src/
│   ├── __init__.py
│   ├── handler.py          # Main Lambda handler
│   ├── models.py           # Data models
│   ├── validators.py       # Input validation
│   └── services/
│       ├── __init__.py
│       ├── dynamodb.py     # DynamoDB operations
│       └── notification.py # SNS operations
├── tests/
│   ├── __init__.py
│   ├── test_handler.py
│   └── test_validators.py
├── events/
│   └── api_gateway_event.json
├── template.yaml           # SAM template
├── requirements.txt
└── README.md

Step 1: Create the Project

# Create project directory
mkdir user-registration-api
cd user-registration-api

# Create directory structure
mkdir -p src/services tests events

# Create __init__.py files
touch src/__init__.py src/services/__init__.py tests/__init__.py

Step 2: Define Requirements

Create requirements.txt:

boto3==1.34.34
pydantic==2.5.3
python-json-logger==2.0.7

Why these dependencies?

• boto3: AWS SDK (already in Lambda, but pinned for local testing)

• pydantic: Data validation (learned this after dealing with bad input)

• python-json-logger: Structured logging (essential for debugging)

Step 3: Build the Data Models

Create src/models.py:

from pydantic import BaseModel, EmailStr, Field, validator
from datetime import datetime
from typing import Optional
import re


class UserRegistration(BaseModel):
    """User registration data model with validation"""
    
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=30)
    full_name: str = Field(..., min_length=2, max_length=100)
    phone: Optional[str] = None
    
    @validator('username')
    def validate_username(cls, v):
        """Ensure username contains only alphanumeric and underscores"""
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('Username must contain only letters, numbers, and underscores')
        return v.lower()
    
    @validator('phone')
    def validate_phone(cls, v):
        """Basic phone validation"""
        if v and not re.match(r'^\+?1?\d{10,15}$', v):
            raise ValueError('Invalid phone number format')
        return v


class UserRecord(BaseModel):
    """DynamoDB user record"""
    
    user_id: str
    email: EmailStr
    username: str
    full_name: str
    phone: Optional[str] = None
    created_at: str
    status: str = 'pending'
    
    class Config:
        """Pydantic configuration"""
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

Step 4: Create Validators

Create src/validators.py:

import json
from typing import Dict, Tuple
from pydantic import ValidationError
from src.models import UserRegistration


def validate_request_body(event: Dict) -> Tuple[bool, any]:
    """
    Validate API Gateway request body
    
    Returns:
        Tuple of (is_valid, data_or_errors)
    """
    try:
        # Parse JSON body
        body = event.get('body', '{}')
        if isinstance(body, str):
            body = json.loads(body)
        
        # Validate with Pydantic
        user_data = UserRegistration(**body)
        return True, user_data
        
    except json.JSONDecodeError as e:
        return False, {'error': 'Invalid JSON', 'detail': str(e)}
        
    except ValidationError as e:
        return False, {'error': 'Validation failed', 'detail': e.errors()}
        
    except Exception as e:
        return False, {'error': 'Unexpected error', 'detail': str(e)}

Step 5: DynamoDB Service

Create src/services/dynamodb.py:

import boto3
import os
from datetime import datetime
from typing import Dict
import uuid
from src.models import UserRecord


class DynamoDBService:
    """Handle DynamoDB operations"""
    
    def __init__(self):
        self.dynamodb = boto3.resource('dynamodb')
        self.table_name = os.environ.get('USERS_TABLE', 'Users')
        self.table = self.dynamodb.Table(self.table_name)
    
    def create_user(self, user_data: Dict) -> UserRecord:
        """
        Create new user record
        
        Args:
            user_data: Validated user data
            
        Returns:
            UserRecord with generated ID
        """
        # Generate user record
        user_record = UserRecord(
            user_id=str(uuid.uuid4()),
            email=user_data['email'],
            username=user_data['username'],
            full_name=user_data['full_name'],
            phone=user_data.get('phone'),
            created_at=datetime.utcnow().isoformat(),
            status='pending'
        )
        
        # Store in DynamoDB
        self.table.put_item(
            Item=user_record.dict(),
            ConditionExpression='attribute_not_exists(email)'  # Prevent duplicates
        )
        
        return user_record
    
    def get_user_by_email(self, email: str) -> Dict | None:
        """Retrieve user by email"""
        response = self.table.query(
            IndexName='EmailIndex',
            KeyConditionExpression='email = :email',
            ExpressionAttributeValues={':email': email}
        )
        
        items = response.get('Items', [])
        return items[0] if items else None

Step 6: Notification Service

Create src/services/notification.py:

import boto3
import os
import json
from typing import Dict


class NotificationService:
    """Handle SNS notifications"""
    
    def __init__(self):
        self.sns = boto3.client('sns')
        self.topic_arn = os.environ.get('SNS_TOPIC_ARN')
    
    def send_registration_notification(self, user_data: Dict) -> str:
        """
        Send user registration notification
        
        Returns:
            Message ID from SNS
        """
        message = {
            'event': 'user_registration',
            'user_id': user_data['user_id'],
            'email': user_data['email'],
            'username': user_data['username'],
            'timestamp': user_data['created_at']
        }
        
        response = self.sns.publish(
            TopicArn=self.topic_arn,
            Message=json.dumps(message),
            Subject=f"New User Registration: {user_data['username']}",
            MessageAttributes={
                'event_type': {
                    'DataType': 'String',
                    'StringValue': 'user_registration'
                }
            }
        )
        
        return response['MessageId']

Step 7: Main Handler

Create src/handler.py:

import json
import logging
from pythonjsonlogger import jsonlogger
from typing import Dict
from src.validators import validate_request_body
from src.services.dynamodb import DynamoDBService
from src.services.notification import NotificationService


# Configure structured logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)

# Initialize services (outside handler for reuse)
db_service = DynamoDBService()
notification_service = NotificationService()


def lambda_handler(event: Dict, context) -> Dict:
    """
    Handle user registration requests
    
    Args:
        event: API Gateway event
        context: Lambda context
        
    Returns:
        API Gateway response
    """
    request_id = context.request_id
    
    logger.info('Processing registration request', extra={
        'request_id': request_id,
        'function': context.function_name
    })
    
    try:
        # Validate request
        is_valid, result = validate_request_body(event)
        
        if not is_valid:
            logger.warning('Validation failed', extra={
                'request_id': request_id,
                'errors': result
            })
            return create_response(400, result)
        
        # Check if user exists
        existing_user = db_service.get_user_by_email(result.email)
        if existing_user:
            logger.info('User already exists', extra={
                'request_id': request_id,
                'email': result.email
            })
            return create_response(409, {
                'error': 'User already exists',
                'email': result.email
            })
        
        # Create user
        user_record = db_service.create_user(result.dict())
        logger.info('User created successfully', extra={
            'request_id': request_id,
            'user_id': user_record.user_id
        })
        
        # Send notification
        message_id = notification_service.send_registration_notification(
            user_record.dict()
        )
        logger.info('Notification sent', extra={
            'request_id': request_id,
            'message_id': message_id
        })
        
        # Return success response
        return create_response(201, {
            'message': 'User registered successfully',
            'user': {
                'user_id': user_record.user_id,
                'username': user_record.username,
                'email': user_record.email
            }
        })
        
    except Exception as e:
        logger.error('Registration failed', extra={
            'request_id': request_id,
            'error': str(e)
        }, exc_info=True)
        
        return create_response(500, {
            'error': 'Internal server error',
            'request_id': request_id
        })


def create_response(status_code: int, body: Dict) -> Dict:
    """Create API Gateway response"""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Headers': 'Content-Type',
            'Access-Control-Allow-Methods': 'POST,OPTIONS'
        },
        'body': json.dumps(body)
    }

Complete Request Flow

Here's how the entire system works together:

Step 8: SAM Template

Create template.yaml:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: User Registration API

Globals:
  Function:
    Timeout: 30
    Runtime: python3.12
    MemorySize: 512
    Environment:
      Variables:
        LOG_LEVEL: INFO

Resources:
  UserRegistrationFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: handler.lambda_handler
      Description: Handle user registration
      Environment:
        Variables:
          USERS_TABLE: !Ref UsersTable
          SNS_TOPIC_ARN: !Ref UserNotificationTopic
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UsersTable
        - SNSPublishMessagePolicy:
            TopicName: !GetAtt UserNotificationTopic.TopicName
      Events:
        RegisterUser:
          Type: Api
          Properties:
            Path: /users/register
            Method: POST

  UsersTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: Users
      AttributeDefinitions:
        - AttributeName: user_id
          AttributeType: S
        - AttributeName: email
          AttributeType: S
      KeySchema:
        - AttributeName: user_id
          KeyType: HASH
      GlobalSecondaryIndexes:
        - IndexName: EmailIndex
          KeySchema:
            - AttributeName: email
              KeyType: HASH
          Projection:
            ProjectionType: ALL
          ProvisionedThroughput:
            ReadCapacityUnits: 5
            WriteCapacityUnits: 5
      BillingMode: PROVISIONED
      ProvisionedThroughput:
        ReadCapacityUnits: 5
        WriteCapacityUnits: 5

  UserNotificationTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: UserRegistrationNotifications
      DisplayName: User Registration Notifications

Outputs:
  ApiEndpoint:
    Description: API Gateway endpoint URL
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/users/register"
    
  UserTableName:
    Description: DynamoDB table name
    Value: !Ref UsersTable
    
  SNSTopicArn:
    Description: SNS Topic ARN
    Value: !Ref UserNotificationTopic

Step 9: Create Test Event

Create events/api_gateway_event.json:

{
  "resource": "/users/register",
  "path": "/users/register",
  "httpMethod": "POST",
  "headers": {
    "Content-Type": "application/json"
  },
  "body": "{\"email\": \"[email protected]\", \"username\": \"johndoe\", \"full_name\": \"John Doe\", \"phone\": \"[phone]\"}"
}

Step 10: Local Testing

Build and Test Locally

# Build the function
sam build

# Invoke locally
sam local invoke UserRegistrationFunction \
  --event events/api_gateway_event.json

# Start local API
sam local start-api

Test with curl

# Test the local endpoint
curl -X POST http://localhost:3000/users/register \
  -H "Content-Type: application/json" \
  -d '{
    "email": "[email protected]",
    "username": "testuser",
    "full_name": "Test User",
    "phone": "[phone]"
  }'

Step 11: Deploy to AWS

# Deploy with SAM
sam deploy --guided

# Follow prompts:
# - Stack Name: user-registration-api
# - AWS Region: us-east-1
# - Confirm changes: Y
# - Allow SAM CLI IAM role creation: Y
# - Save arguments to config: Y

Deployment Flow

Step 12: Test Production Endpoint

# Get the API endpoint from outputs
API_URL=$(aws cloudformation describe-stacks \
  --stack-name user-registration-api \
  --query 'Stacks[0].Outputs[?OutputKey==`ApiEndpoint`].OutputValue' \
  --output text)

# Test registration
curl -X POST $API_URL \
  -H "Content-Type: application/json" \
  -d '{
    "email": "[email protected]",
    "username": "produser",
    "full_name": "Production User"
  }'

Monitoring Your Function

View Logs

# Tail logs in real-time
sam logs -n UserRegistrationFunction --stack-name user-registration-api --tail

# Or use AWS CLI
aws logs tail /aws/lambda/user-registration-api-UserRegistrationFunction --follow

Check Metrics in CloudWatch

Key metrics I monitor:

• Invocations: Total number of calls

• Duration: Execution time

• Errors: Failed invocations

• Throttles: Rate-limited requests

• Concurrent Executions: Number of simultaneous runs

Common Issues I've Encountered

Issue 1: Cold Start Latency

Problem: First request takes 2-3 seconds

Solution:

• Use provisioned concurrency for critical APIs

• Optimize dependencies

• Keep deployment package small

Issue 2: Timeout Errors

Problem: Function times out processing large payloads

Solution:

• Increase timeout in template.yaml

• Break down into smaller operations

• Use async processing for long tasks

Issue 3: Permission Errors

Problem: AccessDeniedException when accessing DynamoDB

Solution:

• Verify IAM policies in SAM template

• Check resource names match exactly

• Review CloudWatch logs for specific permission issues

Key Takeaways

From building this function:

1. Structure Matters: Organized code is easier to test and maintain

2. Validate Early: Catch bad input before processing

3. Log Everything: Structured logs are invaluable for debugging

4. Test Locally First: SAM local saves time and AWS costs

5. Use Infrastructure as Code: SAM templates are reproducible and version-controlled

What's Next?

In Part 4: Event-Driven Architecture with Lambda, we'll explore:

• Multiple event sources (S3, DynamoDB Streams, SQS)

• Asynchronous processing patterns

• Event routing and filtering

• Building a complete event-driven workflow

We'll expand on this registration system to include email verification and user onboarding!

---

This article is based on a real user registration system I built and deployed to production.