API Gateway + Lambda: REST APIs at Scale
Build production-ready REST APIs with AWS API Gateway and Lambda. Learn authentication, rate limiting, caching, custom domains, and cost optimization strategies.
API Gateway + Lambda is the go-to combination for serverless REST APIs. It scales automatically, costs nothing when idle, and requires zero server management. But there are gotchas — cold starts, payload limits, and cost traps. Let’s build it right.
API Gateway Types
| Feature | HTTP API | REST API |
|---|---|---|
| Cost | ~70% cheaper | Full featured |
| Latency | Lower | Higher |
| Auth | JWT, IAM | JWT, IAM, Cognito, Lambda |
| Caching | No | Yes |
| Request validation | No | Yes |
| Usage plans | No | Yes |
| WebSocket | No | Separate product |
Rule of thumb: Start with HTTP API, upgrade to REST API when you need caching or usage plans.
HTTP API with Lambda
Terraform Setup
# HTTP API
resource "aws_apigatewayv2_api" "main" {
name = "my-api"
protocol_type = "HTTP"
description = "Main application API"
cors_configuration {
allow_origins = ["https://myapp.com", "https://staging.myapp.com"]
allow_methods = ["GET", "POST", "PUT", "DELETE", "OPTIONS"]
allow_headers = ["Content-Type", "Authorization", "X-Request-ID"]
expose_headers = ["X-Request-ID"]
max_age = 3600
allow_credentials = true
}
tags = var.tags
}
# Stage
resource "aws_apigatewayv2_stage" "main" {
api_id = aws_apigatewayv2_api.main.id
name = "$default"
auto_deploy = true
access_log_settings {
destination_arn = aws_cloudwatch_log_group.api.arn
format = jsonencode({
requestId = "$context.requestId"
ip = "$context.identity.sourceIp"
requestTime = "$context.requestTime"
httpMethod = "$context.httpMethod"
routeKey = "$context.routeKey"
status = "$context.status"
responseLength = "$context.responseLength"
latency = "$context.responseLatency"
integrationLatency = "$context.integrationLatency"
})
}
default_route_settings {
throttling_burst_limit = 1000
throttling_rate_limit = 500
}
tags = var.tags
}
# Lambda integration
resource "aws_apigatewayv2_integration" "lambda" {
api_id = aws_apigatewayv2_api.main.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_function.api.invoke_arn
payload_format_version = "2.0"
timeout_milliseconds = 30000
}
# Routes
resource "aws_apigatewayv2_route" "get_items" {
api_id = aws_apigatewayv2_api.main.id
route_key = "GET /items"
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
authorization_type = "JWT"
authorizer_id = aws_apigatewayv2_authorizer.jwt.id
}
resource "aws_apigatewayv2_route" "get_item" {
api_id = aws_apigatewayv2_api.main.id
route_key = "GET /items/{id}"
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
authorization_type = "JWT"
authorizer_id = aws_apigatewayv2_authorizer.jwt.id
}
resource "aws_apigatewayv2_route" "create_item" {
api_id = aws_apigatewayv2_api.main.id
route_key = "POST /items"
target = "integrations/${aws_apigatewayv2_integration.lambda.id}"
authorization_type = "JWT"
authorizer_id = aws_apigatewayv2_authorizer.jwt.id
}
# JWT Authorizer (Cognito)
resource "aws_apigatewayv2_authorizer" "jwt" {
api_id = aws_apigatewayv2_api.main.id
authorizer_type = "JWT"
name = "cognito-jwt"
identity_sources = ["$request.header.Authorization"]
jwt_configuration {
audience = [aws_cognito_user_pool_client.api.id]
issuer = "https://${aws_cognito_user_pool.main.endpoint}"
}
}
# Lambda permission
resource "aws_lambda_permission" "api_gateway" {
statement_id = "AllowAPIGateway"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.api.function_name
principal = "apigateway.amazonaws.com"
source_arn = "${aws_apigatewayv2_api.main.execution_arn}/*/*"
}
Lambda Handler
Python with AWS Lambda Powertools
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.event_handler import APIGatewayHttpResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.validation import validate
logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayHttpResolver()
# Initialize outside handler
import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])
@app.get("/items")
@tracer.capture_method
def get_items():
"""List all items with pagination."""
params = app.current_event.query_string_parameters or {}
limit = int(params.get('limit', 20))
cursor = params.get('cursor')
scan_params = {'Limit': limit}
if cursor:
scan_params['ExclusiveStartKey'] = {'pk': cursor}
response = table.scan(**scan_params)
return {
'items': response.get('Items', []),
'next_cursor': response.get('LastEvaluatedKey', {}).get('pk')
}
@app.get("/items/<item_id>")
@tracer.capture_method
def get_item(item_id: str):
"""Get a single item by ID."""
response = table.get_item(Key={'pk': item_id})
if 'Item' not in response:
return {'message': 'Item not found'}, 404
return response['Item']
@app.post("/items")
@tracer.capture_method
def create_item():
"""Create a new item."""
body = app.current_event.json_body
# Validate input
required = ['name', 'description']
for field in required:
if field not in body:
return {'message': f'Missing field: {field}'}, 400
item_id = str(uuid.uuid4())
item = {
'pk': item_id,
'name': body['name'],
'description': body['description'],
'created_at': datetime.utcnow().isoformat(),
'created_by': app.current_event.request_context.authorizer.claims['sub']
}
table.put_item(Item=item)
metrics.add_metric(name="ItemsCreated", unit="Count", value=1)
return item, 201
@app.put("/items/<item_id>")
@tracer.capture_method
def update_item(item_id: str):
"""Update an existing item."""
body = app.current_event.json_body
try:
response = table.update_item(
Key={'pk': item_id},
UpdateExpression='SET #name = :name, description = :desc, updated_at = :ts',
ExpressionAttributeNames={'#name': 'name'},
ExpressionAttributeValues={
':name': body.get('name'),
':desc': body.get('description'),
':ts': datetime.utcnow().isoformat()
},
ConditionExpression='attribute_exists(pk)',
ReturnValues='ALL_NEW'
)
return response['Attributes']
except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
return {'message': 'Item not found'}, 404
@app.delete("/items/<item_id>")
@tracer.capture_method
def delete_item(item_id: str):
"""Delete an item."""
try:
table.delete_item(
Key={'pk': item_id},
ConditionExpression='attribute_exists(pk)'
)
return '', 204
except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
return {'message': 'Item not found'}, 404
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_HTTP)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def handler(event: dict, context: LambdaContext) -> dict:
return app.resolve(event, context)
REST API with Caching
Terraform Configuration
resource "aws_api_gateway_rest_api" "main" {
name = "my-rest-api"
description = "REST API with caching"
endpoint_configuration {
types = ["REGIONAL"]
}
tags = var.tags
}
resource "aws_api_gateway_resource" "items" {
rest_api_id = aws_api_gateway_rest_api.main.id
parent_id = aws_api_gateway_rest_api.main.root_resource_id
path_part = "items"
}
resource "aws_api_gateway_method" "get_items" {
rest_api_id = aws_api_gateway_rest_api.main.id
resource_id = aws_api_gateway_resource.items.id
http_method = "GET"
authorization = "COGNITO_USER_POOLS"
authorizer_id = aws_api_gateway_authorizer.cognito.id
request_parameters = {
"method.request.querystring.category" = false
"method.request.querystring.limit" = false
}
}
resource "aws_api_gateway_integration" "get_items" {
rest_api_id = aws_api_gateway_rest_api.main.id
resource_id = aws_api_gateway_resource.items.id
http_method = aws_api_gateway_method.get_items.http_method
integration_http_method = "POST"
type = "AWS_PROXY"
uri = aws_lambda_function.api.invoke_arn
# Cache key parameters
cache_key_parameters = [
"method.request.querystring.category"
]
}
resource "aws_api_gateway_stage" "prod" {
deployment_id = aws_api_gateway_deployment.main.id
rest_api_id = aws_api_gateway_rest_api.main.id
stage_name = "prod"
# Enable caching
cache_cluster_enabled = true
cache_cluster_size = "0.5" # GB
variables = {
environment = "production"
}
access_log_settings {
destination_arn = aws_cloudwatch_log_group.api.arn
format = file("${path.module}/log-format.json")
}
xray_tracing_enabled = true
tags = var.tags
}
resource "aws_api_gateway_method_settings" "get_items" {
rest_api_id = aws_api_gateway_rest_api.main.id
stage_name = aws_api_gateway_stage.prod.stage_name
method_path = "items/GET"
settings {
caching_enabled = true
cache_ttl_in_seconds = 300
metrics_enabled = true
logging_level = "INFO"
throttling_burst_limit = 1000
throttling_rate_limit = 500
}
}
Custom Domains
Certificate and Domain Setup
# ACM Certificate (must be in us-east-1 for CloudFront)
resource "aws_acm_certificate" "api" {
provider = aws.us_east_1
domain_name = "api.myapp.com"
validation_method = "DNS"
lifecycle {
create_before_destroy = true
}
tags = var.tags
}
resource "aws_route53_record" "cert_validation" {
for_each = {
for dvo in aws_acm_certificate.api.domain_validation_options : dvo.domain_name => {
name = dvo.resource_record_name
record = dvo.resource_record_value
type = dvo.resource_record_type
}
}
zone_id = data.aws_route53_zone.main.zone_id
name = each.value.name
type = each.value.type
records = [each.value.record]
ttl = 60
}
resource "aws_acm_certificate_validation" "api" {
provider = aws.us_east_1
certificate_arn = aws_acm_certificate.api.arn
validation_record_fqdns = [for record in aws_route53_record.cert_validation : record.fqdn]
}
# Custom domain for HTTP API
resource "aws_apigatewayv2_domain_name" "api" {
domain_name = "api.myapp.com"
domain_name_configuration {
certificate_arn = aws_acm_certificate.api.arn
endpoint_type = "REGIONAL"
security_policy = "TLS_1_2"
}
depends_on = [aws_acm_certificate_validation.api]
tags = var.tags
}
resource "aws_apigatewayv2_api_mapping" "api" {
api_id = aws_apigatewayv2_api.main.id
domain_name = aws_apigatewayv2_domain_name.api.id
stage = aws_apigatewayv2_stage.main.id
}
# DNS record
resource "aws_route53_record" "api" {
zone_id = data.aws_route53_zone.main.zone_id
name = "api.myapp.com"
type = "A"
alias {
name = aws_apigatewayv2_domain_name.api.domain_name_configuration[0].target_domain_name
zone_id = aws_apigatewayv2_domain_name.api.domain_name_configuration[0].hosted_zone_id
evaluate_target_health = false
}
}
Rate Limiting with Usage Plans
# API Key
resource "aws_api_gateway_api_key" "partner" {
name = "partner-key"
enabled = true
}
# Usage Plan
resource "aws_api_gateway_usage_plan" "partner" {
name = "partner-plan"
api_stages {
api_id = aws_api_gateway_rest_api.main.id
stage = aws_api_gateway_stage.prod.stage_name
throttle {
path = "/items/GET"
burst_limit = 100
rate_limit = 50
}
}
quota_settings {
limit = 10000
period = "DAY"
}
throttle_settings {
burst_limit = 500
rate_limit = 200
}
}
resource "aws_api_gateway_usage_plan_key" "partner" {
key_id = aws_api_gateway_api_key.partner.id
key_type = "API_KEY"
usage_plan_id = aws_api_gateway_usage_plan.partner.id
}
Cold Start Optimization
Provisioned Concurrency
resource "aws_lambda_alias" "live" {
name = "live"
function_name = aws_lambda_function.api.function_name
function_version = aws_lambda_function.api.version
}
resource "aws_lambda_provisioned_concurrency_config" "api" {
function_name = aws_lambda_function.api.function_name
qualifier = aws_lambda_alias.live.name
provisioned_concurrent_executions = 5
}
# Update API Gateway to use alias
resource "aws_apigatewayv2_integration" "lambda" {
api_id = aws_apigatewayv2_api.main.id
integration_type = "AWS_PROXY"
integration_uri = aws_lambda_alias.live.invoke_arn # Use alias
payload_format_version = "2.0"
}
Scheduled Warmup
resource "aws_cloudwatch_event_rule" "warmup" {
name = "api-warmup"
schedule_expression = "rate(5 minutes)"
}
resource "aws_cloudwatch_event_target" "warmup" {
rule = aws_cloudwatch_event_rule.warmup.name
arn = aws_lambda_function.api.arn
input = jsonencode({
warmup = true
})
}
resource "aws_lambda_permission" "warmup" {
statement_id = "AllowEventBridge"
action = "lambda:InvokeFunction"
function_name = aws_lambda_function.api.function_name
principal = "events.amazonaws.com"
source_arn = aws_cloudwatch_event_rule.warmup.arn
}
Cost Optimization
Cost Breakdown
HTTP API pricing (us-east-1):
- First 300M requests: $1.00/million
- Next 700M: $0.90/million
- Over 1B: $0.80/million
REST API pricing:
- $3.50/million requests
- Cache: $0.02-$3.80/hour depending on size
Lambda pricing:
- $0.20/million requests
- $0.0000166667/GB-second
Cost Tips
cost_optimization:
- Use HTTP API unless you need REST API features
- Enable caching for read-heavy endpoints
- Right-size Lambda memory (affects CPU too)
- Use provisioned concurrency only for latency-critical paths
- Compress responses to reduce data transfer
- Consider CloudFront for global distribution
Key Takeaways
- Start with HTTP API — 70% cheaper, good enough for most use cases
- Use Powertools — logging, tracing, and metrics out of the box
- Cache aggressively — REST API caching saves Lambda invocations
- Custom domains matter — professional APIs need professional URLs
- Monitor cold starts — provision concurrency for latency-sensitive endpoints
“The best API is invisible — users don’t notice how it works, only that it works.”