はじめに
サーバーレスアーキテクチャを使えば、サーバーを管理せずにアプリケーションを構築・実行できます。AWS Lambdaはイベントに応じてコードを実行し、API Gatewayは完全マネージドなHTTPエンドポイントを提供します。DynamoDBと組み合わせれば、スケーラブルでコスト効率の良いAPIスタックが完成します。
このチュートリアルでは、タスク管理システムの完全なCRUD REST APIを構築します。
前提条件
- AWSアカウント（Free Tier対象）
- AWS CLI設定済み（
aws configure） - SAM CLIインストール済み
- Python 3.12以上
- REST APIの基本知識
プロジェクトセットアップ
まず、新しいSAMプロジェクトを作成します：
# SAM CLIをインストール
pip install aws-sam-cli
# プロジェクトを初期化
sam init --runtime python3.12 --name task-api --app-template hello-world
cd task-api
プロジェクト構造
task-api/
├── template.yaml # SAM/CloudFormationテンプレート
├── src/
│ ├── handlers/ # Lambda関数ハンドラー
│ │ ├── create_task.py
│ │ ├── get_task.py
│ │ ├── list_tasks.py
│ │ ├── update_task.py
│ │ └── delete_task.py
│ └── utils/ # 共通ユーティリティ
│ ├── dynamodb.py
│ └── response.py
├── tests/
├── requirements.txt
└── samconfig.toml
ステップ1：SAMでインフラを定義する
template.yamlを編集します：
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: サーバーレスタスク管理API
Globals:
Function:
Timeout: 10
Runtime: python3.12
MemorySize: 256
Environment:
Variables:
TABLE_NAME: !Ref TasksTable
Architectures:
- arm64 # Graviton2 — より安価で高速
Resources:
# DynamoDBテーブル
TasksTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: tasks
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- AttributeName: task_id
AttributeType: S
- AttributeName: user_id
AttributeType: S
KeySchema:
- AttributeName: user_id
KeyType: HASH
- AttributeName: task_id
KeyType: RANGE
# API Gateway
TaskApi:
Type: AWS::Serverless::Api
Properties:
StageName: prod
Cors:
AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
AllowHeaders: "'Content-Type,Authorization'"
AllowOrigin: "'*'"
# Lambda関数（各CRUDエンドポイント）
CreateTaskFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: src/
Handler: handlers.create_task.handler
Events:
CreateTask:
Type: Api
Properties:
RestApiId: !Ref TaskApi
Path: /tasks
Method: POST
Policies:
- DynamoDBCrudPolicy:
TableName: !Ref TasksTable
ステップ2：ユーティリティモジュールを構築する
DynamoDBヘルパー
import os
import boto3
from boto3.dynamodb.conditions import Key
_table = None
def get_table():
"""Lambda実行間でDynamoDB接続を再利用する。"""
global _table
if _table is None:
dynamodb = boto3.resource('dynamodb')
_table = dynamodb.Table(os.environ['TABLE_NAME'])
return _table
def put_item(item: dict) -> dict:
table = get_table()
table.put_item(Item=item)
return item
def query_by_user(user_id: str, limit: int = 50) -> list[dict]:
table = get_table()
response = table.query(
KeyConditionExpression=Key('user_id').eq(user_id),
Limit=limit,
ScanIndexForward=False
)
return response.get('Items', [])
レスポンスヘルパー
import json
from decimal import Decimal
class DecimalEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, Decimal):
return int(obj) if obj % 1 == 0 else float(obj)
return super().default(obj)
def success(body, status_code=200):
return {
'statusCode': status_code,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*',
},
'body': json.dumps(body, cls=DecimalEncoder)
}
def error(message, status_code=400):
return success({'error': message}, status_code)
ステップ3：CRUDハンドラーを実装する
タスク作成
import json
import uuid
from datetime import datetime, timezone
from utils.dynamodb import put_item
from utils.response import success, error
def handler(event, context):
body = json.loads(event.get('body', '{}'))
title = body.get('title', '').strip()
if not title:
return error('title is required')
now = datetime.now(timezone.utc).isoformat()
task = {
'task_id': str(uuid.uuid4()),
'user_id': body.get('user_id', 'default'),
'title': title,
'description': body.get('description', ''),
'status': 'pending',
'priority': body.get('priority', 'medium'),
'created_at': now,
'updated_at': now,
}
put_item(task)
return success(task, 201)
ステップ4：ローカルテスト
# ビルド
sam build
# ローカルAPI起動（Docker必要）
sam local start-api --port 3000
# テスト
curl -X POST http://localhost:3000/tasks \
-H 'Content-Type: application/json' \
-d '{"title": "Lambdaを学ぶ", "priority": "high"}'
curl http://localhost:3000/tasks
ステップ5：ミドルウェアでリクエスト検証を追加
import json
import logging
import traceback
from functools import wraps
from utils.response import error
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def api_handler(func):
"""一貫したエラーハンドリングとロギングのデコレーター。"""
@wraps(func)
def wrapper(event, context):
logger.info(json.dumps({
'method': event.get('httpMethod'),
'path': event.get('path'),
'request_id': context.aws_request_id,
}))
try:
return func(event, context)
except json.JSONDecodeError:
return error('リクエストボディのJSONが不正です', 400)
except Exception as e:
logger.error(traceback.format_exc())
return error('Internal server error', 500)
return wrapper
ステップ6：AWSへデプロイ
# 初回デプロイ（ガイド付き）
sam deploy --guided
# 以降のデプロイ
sam deploy
ステップ7：APIキー認証を追加
TaskApi:
Type: AWS::Serverless::Api
Properties:
StageName: prod
Auth:
ApiKeyRequired: true
UsagePlan:
CreateUsagePlan: PER_API
Throttle:
BurstLimit: 50
RateLimit: 20
Quota:
Limit: 10000
Period: MONTH
コスト見積もり
| サービス | 無料枠 | 無料枠超過後 |
|----------|--------|-------------|
| Lambda | 月100万リクエスト | 100万リクエストあたり$0.20 |
| API Gateway | 月100万コール | 100万コールあたり$3.50 |
| DynamoDB | 25GBストレージ | GB/月あたり$0.25 |
低トラフィックAPI（約1万リクエスト/日）の場合、月額コストは無料枠内で$0、超過後は約$3です。
まとめ
完全なサーバーレスREST APIを構築しました：