Serverlessアーキテクチャパターン完全ガイド2025：Lambda、Step Functions、イベントソーシング、コスト最適化

2014: AWS Lambda リリース（最初のFaaS）
2016: Azure Functions、Google Cloud Functions
2017: AWS Step Functions、SAM リリース
2018: Lambda Layers、ALBサポート
2019: Provisioned Concurrency、RDS Proxy
2020: Lambda Container Image、EventBridge
2021: Lambda URL、Graviton2サポート
2022: Lambda SnapStart（Java）、ストリーミングレスポンス
2023: Lambda Advanced Logging、Step Functions改善
2024: Lambda パフォーマンス最適化、ARM64全面サポート
2025: Lambda最大メモリ10GB、Step Functions Distributed Map強化

# order_create.py - 注文作成のみ担当
import json
import boto3
import os
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['ORDERS_TABLE'])
sns = boto3.client('sns')

def handler(event, context):
    body = json.loads(event['body'])

    order = {
        'orderId': context.aws_request_id,
        'userId': body['userId'],
        'items': body['items'],
        'total': calculate_total(body['items']),
        'status': 'CREATED',
        'createdAt': datetime.utcnow().isoformat()
    }

    table.put_item(Item=order)

    # イベント発行
    sns.publish(
        TopicArn=os.environ['ORDER_TOPIC'],
        Message=json.dumps(order),
        MessageAttributes={
            'eventType': {
                'DataType': 'String',
                'StringValue': 'OrderCreated'
            }
        }
    )

    return {
        'statusCode': 201,
        'body': json.dumps(order)
    }

def calculate_total(items):
    return sum(item['price'] * item['quantity'] for item in items)

// app.ts - モノリシックLambda
import express from 'express';
import serverless from 'serverless-http';

const app = express();
app.use(express.json());

// 複数ルートを一つのLambdaで処理
app.get('/orders', async (req, res) => {
  const orders = await getOrders(req.query);
  res.json(orders);
});

app.post('/orders', async (req, res) => {
  const order = await createOrder(req.body);
  res.status(201).json(order);
});

app.get('/orders/:id', async (req, res) => {
  const order = await getOrder(req.params.id);
  if (!order) return res.status(404).json({ error: 'Not found' });
  res.json(order);
});

app.put('/orders/:id/status', async (req, res) => {
  const order = await updateOrderStatus(req.params.id, req.body.status);
  res.json(order);
});

app.delete('/orders/:id', async (req, res) => {
  await cancelOrder(req.params.id);
  res.status(204).send();
});

export const handler = serverless(app);

# serverless.yml - Fan-outアーキテクチャ
service: order-processing

provider:
  name: aws
  runtime: nodejs20.x

functions:
  orderReceiver:
    handler: src/receiver.handler
    events:
      - http:
          path: /orders
          method: post
    environment:
      FAN_OUT_TOPIC: !Ref OrderFanOutTopic

  inventoryCheck:
    handler: src/inventory.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

  paymentProcess:
    handler: src/payment.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

  notificationSend:
    handler: src/notification.handler
    events:
      - sns:
          arn: !Ref OrderFanOutTopic
          filterPolicy:
            eventType:
              - OrderCreated

resources:
  Resources:
    OrderFanOutTopic:
      Type: AWS::SNS::Topic
      Properties:
        TopicName: order-fan-out

{
  "Comment": "注文処理ワークフロー",
  "StartAt": "ValidateOrder",
  "States": {
    "ValidateOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:validate-order",
      "Next": "CheckInventory",
      "Retry": [
        {
          "ErrorEquals": ["ServiceException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "Next": "OrderFailed"
        }
      ]
    },
    "CheckInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:check-inventory",
      "Next": "ProcessPaymentOrWait"
    },
    "ProcessPaymentOrWait": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.inventoryAvailable",
          "BooleanEquals": true,
          "Next": "ProcessPayment"
        }
      ],
      "Default": "WaitForInventory"
    },
    "WaitForInventory": {
      "Type": "Wait",
      "Seconds": 300,
      "Next": "CheckInventory"
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:process-payment",
      "Next": "ParallelFulfillment"
    },
    "ParallelFulfillment": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "UpdateDatabase",
          "States": {
            "UpdateDatabase": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:update-db",
              "End": true
            }
          }
        },
        {
          "StartAt": "SendNotification",
          "States": {
            "SendNotification": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:send-notification",
              "End": true
            }
          }
        },
        {
          "StartAt": "InitiateShipping",
          "States": {
            "InitiateShipping": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:initiate-shipping",
              "End": true
            }
          }
        }
      ],
      "Next": "OrderCompleted"
    },
    "OrderCompleted": {
      "Type": "Succeed"
    },
    "OrderFailed": {
      "Type": "Fail",
      "Error": "OrderProcessingFailed",
      "Cause": "Order validation or processing failed"
    }
  }
}

# callback_handler.py - 人間の承認を待つLambda
import json
import boto3

sfn = boto3.client('stepfunctions')
ses = boto3.client('ses')

def request_approval(event, context):
    """Step Functionsがタスクトークンと共に呼び出す"""
    task_token = event['taskToken']
    order = event['order']

    # 承認リンク付きメール送信
    approval_url = f"https://api.example.com/approve?token={task_token}"
    reject_url = f"https://api.example.com/reject?token={task_token}"

    ses.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': ['manager@example.com']},
        Message={
            'Subject': {'Data': f"注文承認リクエスト: {order['orderId']}"},
            'Body': {
                'Html': {
                    'Data': f"""
                    <h2>注文承認リクエスト</h2>
                    <p>注文ID: {order['orderId']}</p>
                    <p>金額: {order['total']}円</p>
                    <a href="{approval_url}">承認</a> |
                    <a href="{reject_url}">拒否</a>
                    """
                }
            }
        }
    )

def handle_approval(event, context):
    """承認/拒否コールバック処理"""
    params = event['queryStringParameters']
    task_token = params['token']

    if 'approve' in event['path']:
        sfn.send_task_success(
            taskToken=task_token,
            output=json.dumps({'approved': True})
        )
    else:
        sfn.send_task_failure(
            taskToken=task_token,
            error='Rejected',
            cause='Manager rejected the order'
        )

    return {
        'statusCode': 200,
        'body': json.dumps({'message': '処理完了'})
    }

# event_store.py
import json
import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
event_store = dynamodb.Table('EventStore')
sns = boto3.client('sns')

def append_event(aggregate_id, event_type, data, version):
    """イベントを保存・発行"""
    event = {
        'aggregateId': aggregate_id,
        'version': version,
        'eventType': event_type,
        'data': data,
        'timestamp': datetime.utcnow().isoformat(),
        'metadata': {
            'correlationId': data.get('correlationId', ''),
            'causationId': data.get('causationId', '')
        }
    }

    # 楽観的ロック：versionが既に存在すれば失敗
    event_store.put_item(
        Item=event,
        ConditionExpression='attribute_not_exists(version)'
    )

    # イベント発行
    sns.publish(
        TopicArn='arn:aws:sns:ap-northeast-1:123456789:domain-events',
        Message=json.dumps(event),
        MessageAttributes={
            'eventType': {
                'DataType': 'String',
                'StringValue': event_type
            }
        }
    )

    return event

def replay_events(aggregate_id):
    """特定Aggregateの全イベントを再生"""
    response = event_store.query(
        KeyConditionExpression='aggregateId = :aid',
        ExpressionAttributeValues={':aid': aggregate_id},
        ScanIndexForward=True
    )
    return response['Items']

{
  "Comment": "注文Saga - 補償トランザクション付き",
  "StartAt": "ReserveInventory",
  "States": {
    "ReserveInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:reserve-inventory",
      "Next": "ProcessPayment",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "InventoryReservationFailed"
      }]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:process-payment",
      "Next": "ConfirmOrder",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RollbackInventory"
      }]
    },
    "ConfirmOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:confirm-order",
      "Next": "SagaCompleted",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RollbackPayment"
      }]
    },
    "RollbackPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:rollback-payment",
      "Next": "RollbackInventory"
    },
    "RollbackInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789:function:rollback-inventory",
      "Next": "SagaFailed"
    },
    "InventoryReservationFailed": {
      "Type": "Fail",
      "Error": "InventoryReservationFailed",
      "Cause": "Could not reserve inventory"
    },
    "SagaCompleted": {
      "Type": "Succeed"
    },
    "SagaFailed": {
      "Type": "Fail",
      "Error": "SagaFailed",
      "Cause": "Order saga failed, all compensations executed"
    }
  }
}

リクエスト到着
  |
  v
[実行環境あり？] --No--> [コールドスタート経路]
  |                          |
  Yes                   1. 実行環境プロビジョニング
  |                   2. コードダウンロード（S3）
  v                   3. ランタイム初期化
[ウォームスタート]     4. ハンドラー外コード実行
  |                   5. ハンドラー実行
  v                        |
[ハンドラー実行]           v
  |                   [レスポンス返却]
  v
[レスポンス返却]

# 最適化されたLambda関数構造
import json
import os

# ハンドラー外で初期化（再利用される）
# 1. 接続はグローバルで初期化
import boto3
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['TABLE_NAME'])

# 2. 不要なimportを削除
# BAD: import pandas  （パッケージサイズ増加）
# GOOD: 必要なものだけimport

# 3. SDK設定最適化
from botocore.config import Config
config = Config(
    connect_timeout=5,
    read_timeout=5,
    retries={'max_attempts': 2}
)
s3 = boto3.client('s3', config=config)

def handler(event, context):
    """ハンドラーは可能な限り軽量に"""
    order_id = event['pathParameters']['orderId']

    response = table.get_item(Key={'orderId': order_id})
    item = response.get('Item')

    if not item:
        return {'statusCode': 404, 'body': json.dumps({'error': 'Not found'})}

    return {'statusCode': 200, 'body': json.dumps(item)}

# SAMテンプレート - Provisioned Concurrency設定
Resources:
  MyFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: app.handler
      Runtime: python3.12
      MemorySize: 512
      AutoPublishAlias: live
      ProvisionedConcurrencyConfig:
        ProvisionedConcurrentExecutions: 10

  # 時間帯別自動スケーリング
  ScalingTarget:
    Type: AWS::ApplicationAutoScaling::ScalableTarget
    Properties:
      MaxCapacity: 100
      MinCapacity: 5
      ResourceId: !Sub function:${MyFunction}:live
      ScalableDimension: lambda:function:ProvisionedConcurrency
      ServiceNamespace: lambda

  ScalingPolicy:
    Type: AWS::ApplicationAutoScaling::ScalingPolicy
    Properties:
      PolicyName: UtilizationScaling
      PolicyType: TargetTrackingScaling
      ScalableTargetId: !Ref ScalingTarget
      TargetTrackingScalingPolicyConfiguration:
        TargetValue: 0.7
        PredefinedMetricSpecification:
          PredefinedMetricType: LambdaProvisionedConcurrencyUtilization

// SnapStart最適化されたJava Lambda
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import org.crac.Core;
import org.crac.Resource;

public class OrderHandler implements RequestHandler<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent>,
                                     Resource {

    private final DynamoDbClient dynamoDb;
    private final ObjectMapper objectMapper;

    public OrderHandler() {
        // SnapStart: この初期化コードはスナップショットに含まれる
        this.dynamoDb = DynamoDbClient.create();
        this.objectMapper = new ObjectMapper();
        Core.getGlobalContext().register(this);
    }

    @Override
    public void beforeCheckpoint(org.crac.Context<? extends Resource> context) {
        // スナップショット前：接続クリーンアップ
    }

    @Override
    public void afterRestore(org.crac.Context<? extends Resource> context) {
        // 復元後：接続再確立
        // 一意性保証（乱数シード再設定など）
    }

    @Override
    public APIGatewayProxyResponseEvent handleRequest(
            APIGatewayProxyRequestEvent event, Context context) {
        // ビジネスロジック
        return new APIGatewayProxyResponseEvent()
            .withStatusCode(200)
            .withBody("{\"message\": \"OK\"}");
    }
}

# SAMテンプレート - REST API
Resources:
  OrdersApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod
      Auth:
        DefaultAuthorizer: CognitoAuthorizer
        Authorizers:
          CognitoAuthorizer:
            UserPoolArn: !GetAtt UserPool.Arn
      MethodSettings:
        - HttpMethod: '*'
          ResourcePath: '/*'
          ThrottlingBurstLimit: 100
          ThrottlingRateLimit: 50
      Cors:
        AllowMethods: "'GET,POST,PUT,DELETE,OPTIONS'"
        AllowHeaders: "'Content-Type,Authorization'"
        AllowOrigin: "'https://example.com'"

  GetOrderFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: src/orders/get.handler
      Runtime: nodejs20.x
      Events:
        GetOrder:
          Type: Api
          Properties:
            RestApiId: !Ref OrdersApi
            Path: /orders/{orderId}
            Method: get

# AppSyncスキーマ
type Order {
  orderId: ID!
  userId: String!
  items: [OrderItem!]!
  total: Float!
  status: OrderStatus!
  createdAt: AWSDateTime!
}

type OrderItem {
  productId: String!
  name: String!
  quantity: Int!
  price: Float!
}

enum OrderStatus {
  CREATED
  PAID
  SHIPPED
  DELIVERED
  CANCELLED
}

type Query {
  getOrder(orderId: ID!): Order
  listOrders(userId: String!, limit: Int, nextToken: String): OrderConnection!
}

type Mutation {
  createOrder(input: CreateOrderInput!): Order!
  updateOrderStatus(orderId: ID!, status: OrderStatus!): Order!
}

type Subscription {
  onOrderStatusChanged(orderId: ID!): Order
    @aws_subscribe(mutations: ["updateOrderStatus"])
}

# websocket_handler.py
import json
import boto3
import os

dynamodb = boto3.resource('dynamodb')
connections_table = dynamodb.Table(os.environ['CONNECTIONS_TABLE'])

def connect(event, context):
    """WebSocket接続"""
    connection_id = event['requestContext']['connectionId']
    user_id = event['requestContext']['authorizer']['userId']

    connections_table.put_item(Item={
        'connectionId': connection_id,
        'userId': user_id
    })

    return {'statusCode': 200}

def disconnect(event, context):
    """WebSocket切断"""
    connection_id = event['requestContext']['connectionId']
    connections_table.delete_item(Key={'connectionId': connection_id})
    return {'statusCode': 200}

def send_message(event, context):
    """メッセージ送信"""
    domain = event['requestContext']['domainName']
    stage = event['requestContext']['stage']
    body = json.loads(event['body'])

    apigw = boto3.client(
        'apigatewaymanagementapi',
        endpoint_url=f'https://{domain}/{stage}'
    )

    # 全接続にブロードキャスト
    connections = connections_table.scan()['Items']
    for conn in connections:
        try:
            apigw.post_to_connection(
                ConnectionId=conn['connectionId'],
                Data=json.dumps(body['message']).encode()
            )
        except apigw.exceptions.GoneException:
            connections_table.delete_item(
                Key={'connectionId': conn['connectionId']}
            )

    return {'statusCode': 200}

# DynamoDB Single Table Design
# PK/SKパターンで複数エンティティを一つのテーブルに格納

ENTITY_PATTERNS = {
    'User': {
        'PK': 'USER#user_id',
        'SK': 'PROFILE'
    },
    'Order': {
        'PK': 'USER#user_id',
        'SK': 'ORDER#order_id'
    },
    'OrderItem': {
        'PK': 'ORDER#order_id',
        'SK': 'ITEM#item_id'
    },
    'Product': {
        'PK': 'PRODUCT#product_id',
        'SK': 'DETAIL'
    }
}

# アクセスパターン別クエリ
def get_user_with_orders(user_id):
    """ユーザーと注文リストを一度に取得"""
    response = table.query(
        KeyConditionExpression='PK = :pk',
        ExpressionAttributeValues={':pk': f'USER#{user_id}'}
    )
    user = None
    orders = []
    for item in response['Items']:
        if item['SK'] == 'PROFILE':
            user = item
        elif item['SK'].startswith('ORDER#'):
            orders.append(item)
    return {'user': user, 'orders': orders}

# S3イベント -> Lambda -> DynamoDB パイプライン
import json
import boto3
import csv
import io

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('ProcessedData')

def process_csv_upload(event, context):
    """S3にアップロードされたCSVを処理"""
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    # S3からファイル読み込み
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response['Body'].read().decode('utf-8')

    # CSVパース・バッチ書き込み
    reader = csv.DictReader(io.StringIO(content))

    with table.batch_writer() as batch:
        for row in reader:
            batch.put_item(Item={
                'id': row['id'],
                'data': row,
                'sourceFile': key
            })

    return {
        'statusCode': 200,
        'processedFile': key
    }

メッセージング選択フロー：
1. リアルタイムストリーミングが必要？
   -> Yes: Kinesis Data Streams
   -> No: 次へ

2. 複数コンシューマーに同時配信？
   -> Yes: 次へ
   -> No: SQS（シンプルキュー）

3. 複雑なイベントルーティング/フィルタリング？
   -> Yes: EventBridge
   -> No: SNS

4. イベントリプレイが必要？
   -> Yes: EventBridge（アーカイブ）またはKinesis（保持）
   -> No: SNS/SQS

# Dockerfile - Lambdaコンテナイメージ
FROM public.ecr.aws/lambda/python:3.12

# 依存関係インストール
COPY requirements.txt .
RUN pip install -r requirements.txt

# アプリケーションコード
COPY app/ ./app/

# Lambdaハンドラー指定
CMD ["app.main.handler"]

Lambdaコスト = リクエスト数コスト + 実行時間コスト

リクエスト数コスト：
  - 月100万件無料
  - 以降100万件あたり約0.20 USD

実行時間コスト（x86）：
  - 128MB: 0.0000000021 USD / ms
  - 512MB: 0.0000000083 USD / ms
  - 1024MB: 0.0000000167 USD / ms
  - 1769MB（1 vCPU）: 0.0000000289 USD / ms
  - 10240MB: 0.0000001667 USD / ms

ARM64（Graviton2）価格：
  - x86比約20%安い
  - パフォーマンスは同等以上

# Lambda Powertools - 構造化ロギング、トレーシング、メトリクス
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.event_handler import APIGatewayRestResolver

logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayRestResolver()

@app.get("/orders/<order_id>")
@tracer.capture_method
def get_order(order_id: str):
    logger.info("Fetching order", extra={"order_id": order_id})

    order = fetch_order(order_id)

    metrics.add_metric(name="OrderFetched", unit=MetricUnit.Count, value=1)
    metrics.add_dimension(name="Environment", value="production")

    return {"order": order}

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def handler(event, context):
    return app.resolve(event, context)

# X-Ray SDKで外部呼び出しをトレース
from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

# 全AWS SDK呼び出しを自動トレース
patch_all()

@xray_recorder.capture('process_order')
def process_order(order):
    subsegment = xray_recorder.begin_subsegment('validate')
    try:
        validate_order(order)
        subsegment.put_annotation('valid', True)
    except Exception as e:
        subsegment.put_annotation('valid', False)
        subsegment.add_exception(e)
        raise
    finally:
        xray_recorder.end_subsegment()

    save_order(order)
    publish_event(order)

Resources:
  # Lambdaエラー率アラーム
  LambdaErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: lambda-high-error-rate
      MetricName: Errors
      Namespace: AWS/Lambda
      Dimensions:
        - Name: FunctionName
          Value: !Ref MyFunction
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 2
      Threshold: 5
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic

  # Lambdaスロットルアラーム
  LambdaThrottleAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: lambda-throttled
      MetricName: Throttles
      Namespace: AWS/Lambda
      Dimensions:
        - Name: FunctionName
          Value: !Ref MyFunction
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 0
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic

# SAM CLIでローカルLambda実行
sam local invoke MyFunction \
  --event events/api-gateway.json \
  --env-vars env.json

# ローカルAPIサーバー起動
sam local start-api --port 3000

# DynamoDB Localと併用
docker run -p 8000:8000 amazon/dynamodb-local
sam local invoke --docker-network host

# test_integration.py
import boto3
import pytest
import json

STACK_NAME = 'my-serverless-app'
API_URL = None

@pytest.fixture(scope='session', autouse=True)
def setup():
    """CloudFormationスタックからAPI URLを取得"""
    global API_URL
    cfn = boto3.client('cloudformation')
    response = cfn.describe_stacks(StackName=STACK_NAME)
    outputs = response['Stacks'][0]['Outputs']
    API_URL = next(o['OutputValue'] for o in outputs if o['OutputKey'] == 'ApiUrl')

def test_create_order():
    """注文作成統合テスト"""
    import requests

    response = requests.post(
        f'{API_URL}/orders',
        json={
            'userId': 'test-user',
            'items': [
                {'productId': 'p1', 'name': 'Widget', 'quantity': 2, 'price': 1000}
            ]
        },
        headers={'Authorization': f'Bearer {get_test_token()}'}
    )

    assert response.status_code == 201
    data = response.json()
    assert 'orderId' in data
    assert data['status'] == 'CREATED'
    assert data['total'] == 2000

# test_unit.py
import json
import pytest
from unittest.mock import MagicMock
from moto import mock_dynamodb, mock_sns

@mock_dynamodb
@mock_sns
def test_create_order_handler():
    """Lambdaハンドラーユニットテスト"""
    import boto3

    dynamodb = boto3.resource('dynamodb', region_name='ap-northeast-1')
    table = dynamodb.create_table(
        TableName='orders',
        KeySchema=[{'AttributeName': 'orderId', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'orderId', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )

    sns = boto3.client('sns', region_name='ap-northeast-1')
    topic = sns.create_topic(Name='order-events')

    import os
    os.environ['ORDERS_TABLE'] = 'orders'
    os.environ['ORDER_TOPIC'] = topic['TopicArn']

    from src.orders.create import handler

    event = {
        'body': json.dumps({
            'userId': 'user123',
            'items': [{'productId': 'p1', 'name': 'Test', 'quantity': 1, 'price': 1000}]
        })
    }

    context = MagicMock()
    context.aws_request_id = 'test-request-id'

    response = handler(event, context)

    assert response['statusCode'] == 201
    body = json.loads(response['body'])
    assert body['userId'] == 'user123'
    assert body['total'] == 1000

クライアント
  |
  v
[API Gateway] --> [Lambda: 注文作成]
                     |
                     v
                  [DynamoDB: 注文保存]
                     |
                     v
                  [EventBridge: OrderCreated発行]
                     |
          +----------+----------+
          |          |          |
          v          v          v
     [Lambda:    [Lambda:    [Lambda:
      在庫確認]   決済処理]   通知送信]
          |          |
          v          v
     [DynamoDB]  [Stripe API]
          |
          v
     [Step Functions: 配送ワークフロー]

[S3: 元ファイルアップロード]
  |
  v
[EventBridge: S3イベント]
  |
  v
[Step Functions: メディアパイプライン]
  |
  +-> [Lambda: メタデータ抽出]
  |
  +-> [Lambda: サムネイル生成]
  |
  +-> [Lambda: ビデオトランスコード開始]
  |      |
  |      v
  |   [MediaConvert]
  |      |
  |      v
  |   [Lambda: トランスコード完了処理]
  |
  +-> [Lambda: AIタグ付け（Rekognition）]
  |
  v
[DynamoDB: メタデータ保存]
  |
  v
[CloudFront: CDN配信]