Go back

Deploying FastAPI Applications on AWS Lambda with Mangum - A Complete Guide

Building serverless APIs has become increasingly popular, and combining FastAPI with AWS Lambda provides a powerful, scalable solution. In this comprehensive guide, we’ll explore how to deploy FastAPI applications on AWS Lambda using Mangum, the ASGI adapter that makes this integration seamless.

Cloud computing and serverless architecture concept
Photo by Pexels

Table of contents

Open Table of contents

What is Mangum?

Mangum is an adapter for running ASGI applications in AWS Lambda. It acts as a bridge between the Lambda runtime environment and ASGI frameworks like FastAPI, allowing you to deploy your FastAPI applications as serverless functions without significant code changes.

Key Benefits

Setting Up Your FastAPI Application

Let’s start by creating a FastAPI application that we’ll deploy to AWS Lambda.

Basic FastAPI App Structure

# app/main.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import List, Optional
import json

app = FastAPI(
    title="AWS Lambda FastAPI",
    description="FastAPI application running on AWS Lambda with Mangum",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Pydantic models
class Item(BaseModel):
    id: Optional[int] = None
    name: str
    description: Optional[str] = None
    price: float
    is_available: bool = True

class ItemResponse(BaseModel):
    id: int
    name: str
    description: Optional[str]
    price: float
    is_available: bool

# In-memory storage (use DynamoDB in production)
items_db = []
next_id = 1

@app.get("/")
async def root():
    return {
        "message": "FastAPI on AWS Lambda with Mangum",
        "version": "1.0.0",
        "endpoints": {
            "docs": "/docs",
            "items": "/items"
        }
    }

@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "fastapi-lambda"}

@app.get("/items", response_model=List[ItemResponse])
async def get_items():
    return items_db

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
    item = next((item for item in items_db if item["id"] == item_id), None)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(item: Item):
    global next_id
    new_item = {
        "id": next_id,
        "name": item.name,
        "description": item.description,
        "price": item.price,
        "is_available": item.is_available
    }
    items_db.append(new_item)
    next_id += 1
    return new_item

@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int, item: Item):
    existing_item = next((i for i in items_db if i["id"] == item_id), None)
    if not existing_item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    existing_item.update({
        "name": item.name,
        "description": item.description,
        "price": item.price,
        "is_available": item.is_available
    })
    return existing_item

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    global items_db
    items_db = [item for item in items_db if item["id"] != item_id]
    return {"message": "Item deleted successfully"}

Lambda Handler with Mangum

Create the Lambda handler using Mangum:

# lambda_function.py
from mangum import Mangum
from app.main import app

# Mangum handler
handler = Mangum(app, lifespan="off")

# For local testing
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Dependencies and Configuration

Requirements File

# requirements.txt
fastapi==0.104.1
mangum==0.17.0
pydantic==2.5.0
uvicorn==0.24.0
boto3==1.34.0

Project Structure

fastapi-lambda/
├── app/
│   ├── __init__.py
│   ├── main.py
│   └── models.py
├── lambda_function.py
├── requirements.txt
├── serverless.yml
├── Dockerfile
└── README.md

Deployment Options

Option 1: Using Serverless Framework

Install the Serverless Framework:

npm install -g serverless
npm install serverless-python-requirements

Create a serverless.yml configuration:

# serverless.yml
service: fastapi-lambda-mangum

frameworkVersion: '3'

provider:
  name: aws
  runtime: python3.9
  region: us-east-1
  stage: ${opt:stage, 'dev'}
  environment:
    STAGE: ${self:provider.stage}
  
  iam:
    role:
      statements:
        - Effect: Allow
          Action:
            - dynamodb:Query
            - dynamodb:Scan
            - dynamodb:GetItem
            - dynamodb:PutItem
            - dynamodb:UpdateItem
            - dynamodb:DeleteItem
          Resource:
            - "arn:aws:dynamodb:${self:provider.region}:*:table/${self:service}-${self:provider.stage}-*"

functions:
  api:
    handler: lambda_function.handler
    timeout: 30
    memorySize: 512
    events:
      - httpApi:
          path: /{proxy+}
          method: any
      - httpApi:
          path: /
          method: any

plugins:
  - serverless-python-requirements

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

resources:
  Resources:
    ItemsTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:service}-${self:provider.stage}-items
        BillingMode: PAY_PER_REQUEST
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH

Deploy with:

serverless deploy --stage prod

Option 2: Using AWS SAM

Create a template.yaml for SAM:

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: FastAPI application with Mangum

Globals:
  Function:
    Timeout: 30
    MemorySize: 512
    Runtime: python3.9

Resources:
  FastAPIFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./
      Handler: lambda_function.handler
      Events:
        ApiGateway:
          Type: HttpApi
          Properties:
            Path: /{proxy+}
            Method: any
        RootPath:
          Type: HttpApi
          Properties:
            Path: /
            Method: any
      Environment:
        Variables:
          DYNAMODB_TABLE: !Ref ItemsTable

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

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

Deploy with SAM:

sam build
sam deploy --guided

Option 3: Using Docker and Lambda Container Images

Create a Dockerfile optimized for Lambda:

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

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

# Copy application code
COPY app/ ${LAMBDA_TASK_ROOT}/app/
COPY lambda_function.py ${LAMBDA_TASK_ROOT}

# Set the CMD to your handler
CMD ["lambda_function.handler"]

Build and deploy:

# Build image
docker build -t fastapi-lambda .

# Tag for ECR
docker tag fastapi-lambda:latest <account-id>.dkr.ecr.<region>.amazonaws.com/fastapi-lambda:latest

# Push to ECR
aws ecr get-login-password --region <region> | docker login --username AWS --password-stdin <account-id>.dkr.ecr.<region>.amazonaws.com
docker push <account-id>.dkr.ecr.<region>.amazonaws.com/fastapi-lambda:latest

Integrating with AWS Services

DynamoDB Integration

Enhance your FastAPI app with DynamoDB:

# app/database.py
import boto3
from botocore.exceptions import ClientError
import os
from typing import Dict, List, Optional

class DynamoDBClient:
    def __init__(self):
        self.dynamodb = boto3.resource('dynamodb')
        self.table_name = os.getenv('DYNAMODB_TABLE', 'fastapi-lambda-items')
        self.table = self.dynamodb.Table(self.table_name)
    
    async def create_item(self, item: Dict) -> Dict:
        try:
            self.table.put_item(Item=item)
            return item
        except ClientError as e:
            raise Exception(f"Error creating item: {e}")
    
    async def get_item(self, item_id: str) -> Optional[Dict]:
        try:
            response = self.table.get_item(Key={'id': item_id})
            return response.get('Item')
        except ClientError as e:
            raise Exception(f"Error getting item: {e}")
    
    async def get_all_items(self) -> List[Dict]:
        try:
            response = self.table.scan()
            return response.get('Items', [])
        except ClientError as e:
            raise Exception(f"Error scanning items: {e}")
    
    async def update_item(self, item_id: str, item: Dict) -> Dict:
        try:
            response = self.table.update_item(
                Key={'id': item_id},
                UpdateExpression="SET #name = :name, description = :desc, price = :price, is_available = :available",
                ExpressionAttributeNames={'#name': 'name'},
                ExpressionAttributeValues={
                    ':name': item['name'],
                    ':desc': item.get('description'),
                    ':price': item['price'],
                    ':available': item['is_available']
                },
                ReturnValues="ALL_NEW"
            )
            return response['Attributes']
        except ClientError as e:
            raise Exception(f"Error updating item: {e}")
    
    async def delete_item(self, item_id: str) -> bool:
        try:
            self.table.delete_item(Key={'id': item_id})
            return True
        except ClientError as e:
            raise Exception(f"Error deleting item: {e}")

# Initialize database client
db_client = DynamoDBClient()

Update your main application to use DynamoDB:

# Updated app/main.py (relevant parts)
from app.database import db_client
import uuid

@app.post("/items", response_model=ItemResponse, status_code=201)
async def create_item(item: Item):
    new_item = {
        "id": str(uuid.uuid4()),
        "name": item.name,
        "description": item.description,
        "price": item.price,
        "is_available": item.is_available
    }
    created_item = await db_client.create_item(new_item)
    return created_item

@app.get("/items", response_model=List[ItemResponse])
async def get_items():
    items = await db_client.get_all_items()
    return items

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: str):
    item = await db_client.get_item(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

Best Practices and Optimizations

Cold Start Optimization

# Optimize imports and initialization
import os
import json
from typing import Dict, Any

# Initialize connections outside handler
def get_db_connection():
    # Connection reuse across invocations
    if not hasattr(get_db_connection, '_connection'):
        get_db_connection._connection = boto3.resource('dynamodb')
    return get_db_connection._connection

# Use connection pooling
from mangum import Mangum
from app.main import app

# Configure Mangum for better performance
handler = Mangum(
    app,
    lifespan="off",
    api_gateway_base_path="/",
    text_mime_types=[
        "application/json",
        "application/javascript",
        "application/xml",
        "application/vnd.api+json",
    ]
)

Environment Configuration

# app/config.py
from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    app_name: str = "FastAPI Lambda"
    debug: bool = False
    aws_region: str = "us-east-1"
    dynamodb_table: Optional[str] = None
    cors_origins: list = ["*"]
    
    class Config:
        env_file = ".env"

settings = Settings()

Error Handling and Logging

# app/middleware.py
import logging
import json
from fastapi import Request, Response
from fastapi.middleware.base import BaseHTTPMiddleware
import time

# Configure logging for Lambda
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        
        # Log request
        logger.info(f"Request: {request.method} {request.url}")
        
        response = await call_next(request)
        
        # Log response
        process_time = time.time() - start_time
        logger.info(f"Response: {response.status_code} - {process_time:.4f}s")
        
        return response

# Add to your FastAPI app
app.add_middleware(LoggingMiddleware)

Testing Your Lambda Function

Local Testing

# test_local.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert "FastAPI on AWS Lambda" in response.json()["message"]

def test_health_check():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

def test_create_item():
    item_data = {
        "name": "Test Item",
        "description": "A test item",
        "price": 9.99,
        "is_available": True
    }
    response = client.post("/items", json=item_data)
    assert response.status_code == 201
    assert response.json()["name"] == "Test Item"

def test_get_items():
    response = client.get("/items")
    assert response.status_code == 200
    assert isinstance(response.json(), list)

Lambda Event Testing

# test_lambda.py
import json
from lambda_function import handler

def test_lambda_handler():
    # Sample API Gateway event
    event = {
        "httpMethod": "GET",
        "path": "/health",
        "headers": {},
        "queryStringParameters": None,
        "body": None,
        "isBase64Encoded": False
    }
    
    context = {}
    response = handler(event, context)
    
    assert response["statusCode"] == 200
    body = json.loads(response["body"])
    assert body["status"] == "healthy"

Monitoring and Observability

CloudWatch Integration

# app/monitoring.py
import boto3
import json
from datetime import datetime

cloudwatch = boto3.client('cloudwatch')

def put_custom_metric(metric_name: str, value: float, unit: str = 'Count'):
    """Put custom metric to CloudWatch"""
    try:
        cloudwatch.put_metric_data(
            Namespace='FastAPI/Lambda',
            MetricData=[
                {
                    'MetricName': metric_name,
                    'Value': value,
                    'Unit': unit,
                    'Timestamp': datetime.utcnow()
                }
            ]
        )
    except Exception as e:
        print(f"Error putting metric: {e}")

# Usage in endpoints
@app.post("/items")
async def create_item(item: Item):
    # ... item creation logic
    put_custom_metric('ItemsCreated', 1)
    return created_item

X-Ray Tracing

# Add X-Ray tracing
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

# Patch AWS SDK calls
patch_all()

@xray_recorder.capture('create_item_handler')
async def create_item(item: Item):
    # Your function logic here
    pass

Performance Considerations

Memory and Timeout Configuration

# Serverless configuration for performance
functions:
  api:
    handler: lambda_function.handler
    timeout: 30  # Adjust based on your needs
    memorySize: 1024  # Higher memory = faster CPU
    reservedConcurrency: 50  # Limit concurrent executions
    provisionedConcurrency: 10  # Keep warm instances

Connection Pooling

# app/database.py (optimized)
import boto3
from botocore.config import Config

# Configure connection pooling
config = Config(
    region_name='us-east-1',
    retries={'max_attempts': 3},
    max_pool_connections=50
)

dynamodb = boto3.resource('dynamodb', config=config)

Security Best Practices

IAM Roles and Policies

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:UpdateItem",
        "dynamodb:DeleteItem",
        "dynamodb:Query",
        "dynamodb:Scan"
      ],
      "Resource": "arn:aws:dynamodb:*:*:table/fastapi-lambda-*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:*"
    }
  ]
}

API Authentication

# app/auth.py
from fastapi import HTTPException, Depends, Header
from typing import Optional

async def get_api_key(x_api_key: Optional[str] = Header(None)):
    if not x_api_key or x_api_key != os.getenv("API_KEY"):
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

# Protect endpoints
@app.post("/items", dependencies=[Depends(get_api_key)])
async def create_item(item: Item):
    # Protected endpoint logic
    pass

Troubleshooting Common Issues

Cold Start Problems

  1. Reduce package size: Use only necessary dependencies
  2. Optimize imports: Import only what you need
  3. Use provisioned concurrency: For consistent performance
  4. Connection reuse: Initialize connections outside handlers

Memory Issues

# Memory optimization
import gc
from functools import lru_cache

@lru_cache(maxsize=128)
def cached_expensive_operation(param):
    # Expensive operation with caching
    pass

# Explicit garbage collection
def lambda_handler(event, context):
    try:
        result = handler(event, context)
        return result
    finally:
        gc.collect()  # Force garbage collection

API Gateway Integration Issues

# Handle different event formats
def handler(event, context):
    # Handle both API Gateway v1 and v2 events
    if 'version' in event and event['version'] == '2.0':
        # API Gateway v2 format
        pass
    else:
        # API Gateway v1 format
        pass
    
    return mangum_handler(event, context)

Conclusion

Deploying FastAPI applications on AWS Lambda with Mangum provides a powerful serverless solution that combines the ease of FastAPI development with the scalability and cost-effectiveness of AWS Lambda. By following the patterns and best practices outlined in this guide, you can build robust, scalable APIs that automatically handle traffic spikes while maintaining cost efficiency.

Key takeaways:

Whether you’re building a simple API or a complex microservices architecture, FastAPI with Mangum on AWS Lambda provides a solid foundation for modern serverless applications.

Additional Resources

Share this post on:
Share this post on LinkedInShare this post on FacebookShare this post on XShare this post via TelegramShare this post on PinterestShare this post via email
Previous Post
Beyond Chain-of-Thought - How Brain-Inspired AI Achieves Breakthrough Reasoning with Just 1000 Examples
Next Post
How to create an MCP (Model Context Protocol) server with FastAPI