AWS Lambda with Python: Serverless Computing Guide

The Serverless Paradigm

Serverless computing lets you run code without provisioning or managing servers. AWS Lambda executes your function in response to events, scales automatically from zero to thousands of concurrent executions, and charges you only for the compute time you consume. For many workloads, especially APIs, webhooks, and event-driven processing, serverless is the most cost-effective and operationally simple architecture available.

In this guide, we’ll build a complete serverless REST API using Python, AWS Lambda, API Gateway, and DynamoDB. You’ll learn handler functions, event structures, database operations, and deployment using the AWS Serverless Application Model (SAM).

Your First Lambda Function

A Lambda function is a Python module with a handler function that receives two arguments: event (the input data) and context (runtime information). Let’s start with the basics.

ADVERTISEMENT
import json
import logging

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

def lambda_handler(event, context):
    """Basic Lambda handler that processes API Gateway events."""
    logger.info(f"Received event: {json.dumps(event)}")

    # Extract path and method from API Gateway event
    http_method = event.get("httpMethod", "GET")
    path = event.get("path", "/")
    query_params = event.get("queryStringParameters") or {}
    body = json.loads(event.get("body") or "{}")

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        },
        "body": json.dumps({
            "message": "Hello from Lambda!",
            "method": http_method,
            "path": path,
            "query": query_params
        })
    }

The return format is specific to API Gateway integration. The statusCode, headers, and body fields are required. The body must be a JSON string, not a dictionary.

Building a CRUD API with DynamoDB

DynamoDB is the natural database choice for Lambda: it’s serverless, scales automatically, and has single-digit millisecond latency. Let’s build a complete task management API.

import json
import uuid
import time
import boto3
from decimal import Decimal

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Tasks")


class DecimalEncoder(json.JSONEncoder):
    """Handle Decimal types returned by DynamoDB."""
    def default(self, obj):
        if isinstance(obj, Decimal):
            return int(obj) if obj % 1 == 0 else float(obj)
        return super().default(obj)


def response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps(body, cls=DecimalEncoder),
    }


def create_task(event, context):
    """POST /tasks - Create a new task."""
    body = json.loads(event["body"])

    if not body.get("title"):
        return response(400, {"error": "title is required"})

    task = {
        "id": str(uuid.uuid4()),
        "title": body["title"],
        "description": body.get("description", ""),
        "status": "pending",
        "created_at": int(time.time()),
        "updated_at": int(time.time()),
    }

    table.put_item(Item=task)
    return response(201, task)


def get_task(event, context):
    """GET /tasks/{id} - Get a single task."""
    task_id = event["pathParameters"]["id"]

    result = table.get_item(Key={"id": task_id})
    task = result.get("Item")

    if not task:
        return response(404, {"error": "Task not found"})

    return response(200, task)


def list_tasks(event, context):
    """GET /tasks - List all tasks with optional status filter."""
    query_params = event.get("queryStringParameters") or {}
    status_filter = query_params.get("status")

    if status_filter:
        result = table.scan(
            FilterExpression="task_status = :s",
            ExpressionAttributeValues={":s": status_filter},
        )
    else:
        result = table.scan()

    return response(200, {"tasks": result["Items"], "count": result["Count"]})


def update_task(event, context):
    """PUT /tasks/{id} - Update a task."""
    task_id = event["pathParameters"]["id"]
    body = json.loads(event["body"])

    update_expr = "SET updated_at = :now"
    expr_values = {":now": int(time.time())}

    if "title" in body:
        update_expr += ", title = :title"
        expr_values[":title"] = body["title"]
    if "description" in body:
        update_expr += ", description = :desc"
        expr_values[":desc"] = body["description"]
    if "status" in body:
        update_expr += ", task_status = :status"
        expr_values[":status"] = body["status"]

    result = table.update_item(
        Key={"id": task_id},
        UpdateExpression=update_expr,
        ExpressionAttributeValues=expr_values,
        ReturnValues="ALL_NEW",
    )

    return response(200, result["Attributes"])


def delete_task(event, context):
    """DELETE /tasks/{id} - Delete a task."""
    task_id = event["pathParameters"]["id"]
    table.delete_item(Key={"id": task_id})
    return response(204, {})

SAM Template for Deployment

The AWS Serverless Application Model (SAM) lets you define your entire serverless application as infrastructure-as-code. Save this as template.yaml:

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Runtime: python3.12
    Timeout: 30
    MemorySize: 256
    Environment:
      Variables:
        TABLE_NAME: !Ref TasksTable

Resources:
  CreateTaskFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.create_task
      CodeUri: src/
      Events:
        Api:
          Type: Api
          Properties:
            Path: /tasks
            Method: POST
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref TasksTable

  ListTasksFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.list_tasks
      CodeUri: src/
      Events:
        Api:
          Type: Api
          Properties:
            Path: /tasks
            Method: GET
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref TasksTable

  GetTaskFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.get_task
      CodeUri: src/
      Events:
        Api:
          Type: Api
          Properties:
            Path: /tasks/{id}
            Method: GET
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref TasksTable

  TasksTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: Tasks
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH

Local Testing and Deployment

SAM provides local testing capabilities so you can iterate quickly without deploying to AWS:

# Install SAM CLI
pip install aws-sam-cli

# Invoke a function locally with a test event
sam local invoke CreateTaskFunction -e events/create.json

# Start a local API Gateway
sam local start-api
# Now test with: curl http://localhost:3000/tasks

# Build and deploy
sam build
sam deploy --guided

# After first deploy, subsequent deploys are simpler:
sam build && sam deploy

Production Best Practices

Lambda functions in production need careful attention to cold starts, error handling, and observability. Here are patterns that matter:

import os
import json
import logging
from functools import wraps

logger = logging.getLogger()
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))

# Initialize clients OUTSIDE the handler (reused across invocations)
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])


def handle_errors(func):
    """Decorator for consistent error handling across handlers."""
    @wraps(func)
    def wrapper(event, context):
        try:
            return func(event, context)
        except json.JSONDecodeError:
            return response(400, {"error": "Invalid JSON in request body"})
        except KeyError as e:
            return response(400, {"error": f"Missing required field: {e}"})
        except Exception as e:
            logger.exception("Unhandled exception")
            return response(500, {"error": "Internal server error"})
    return wrapper


@handle_errors
def create_task(event, context):
    body = json.loads(event["body"])
    # ... implementation

Key production tips: initialize SDK clients outside the handler to reuse connections across warm invocations. Set MemorySize to at least 256 MB for Python functions, as CPU scales linearly with memory. Use environment variables for all configuration. Enable AWS X-Ray tracing for distributed tracing across Lambda, API Gateway, and DynamoDB.

Conclusion

You’ve built a complete serverless CRUD API with AWS Lambda, API Gateway, and DynamoDB. The entire stack costs nearly nothing at low traffic and scales automatically to handle thousands of requests per second. Serverless isn’t right for every workload, but for APIs, event processing, and scheduled jobs, it eliminates an enormous amount of operational overhead. From here, explore Lambda Layers for shared dependencies, Step Functions for orchestrating complex workflows, and EventBridge for event-driven architectures.

ADVERTISEMENT

Leave a Comment

Your email address will not be published. Required fields are marked with an asterisk.