# [最適化] ハンドラ外で初期化を行い、Warm 起動時に再利用
import boto3
import os
 
# --- Init Phase (コールドスタート時のみ実行) ---
TABLE_NAME = os.environ["TABLE_NAME"]
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(TABLE_NAME)
# ------------------------------------------------
 
def lambda_handler(event, context):
    """ハンドラは Warm 起動時にも毎回実行される"""
    user_id = event["pathParameters"]["userId"]
 
    response = table.get_item(Key={"userId": user_id})
    return {
        "statusCode": 200,
        "body": json.dumps(response.get("Item", {}))
    }

# [最適化] 遅延インポートで不要なモジュールの初期化を避ける
import json
import os
 
def lambda_handler(event, context):
    action = event.get("action")
 
    if action == "generate_pdf":
        # PDF 生成が必要な場合のみ重いライブラリをインポート
        from reportlab.pdfgen import canvas
        return generate_pdf(event)
    elif action == "send_email":
        import boto3
        ses = boto3.client("ses")
        return send_email(ses, event)
    else:
        return {"statusCode": 400, "body": "Unknown action"}

# [最適化] コネクションプールの再利用パターン
import boto3
import os
from botocore.config import Config
 
# Init Phase: SDK クライアントの設定を最適化
config = Config(
    retries={"max_attempts": 3, "mode": "adaptive"},
    max_pool_connections=10,
    connect_timeout=5,
    read_timeout=10,
)
 
# 各 AWS サービスクライアントを Init Phase で作成
dynamodb_client = boto3.client("dynamodb", config=config)
s3_client = boto3.client("s3", config=config)
sqs_client = boto3.client("sqs", config=config)
 
BUCKET_NAME = os.environ["BUCKET_NAME"]
QUEUE_URL = os.environ["QUEUE_URL"]
 
def lambda_handler(event, context):
    """全クライアントは Warm 起動時に再利用される"""
    # DynamoDB からデータ取得
    item = dynamodb_client.get_item(
        TableName="my-table",
        Key={"pk": {"S": event["id"]}}
    )
 
    # S3 にレポートを保存
    s3_client.put_object(
        Bucket=BUCKET_NAME,
        Key=f"reports/{event['id']}.json",
        Body=json.dumps(item.get("Item", {})),
        ContentType="application/json"
    )
 
    # SQS に通知を送信
    sqs_client.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps({"status": "completed", "id": event["id"]})
    )
 
    return {"statusCode": 200, "body": "Processing complete"}

# Python での軽量パッケージ作成例
# 1. 本番依存のみインストール
pip install -r requirements.txt -t ./package --no-cache-dir
 
# 2. 不要ファイルの除去
cd package
find . -type d -name "__pycache__" -exec rm -rf {} + 2>/dev/null
find . -type d -name "*.dist-info" -exec rm -rf {} + 2>/dev/null
find . -type f -name "*.pyc" -delete 2>/dev/null
find . -type d -name "tests" -exec rm -rf {} + 2>/dev/null
 
# 3. ZIP パッケージの作成
zip -r9 ../function.zip .
cd ..
zip -g function.zip lambda_function.py
 
# 4. サイズ確認
ls -lh function.zip
 
# 5. デプロイ
aws lambda update-function-code \
  --function-name my-function \
  --zip-file fileb://function.zip

# Node.js での軽量パッケージ作成例
# 1. 本番依存のみインストール
npm ci --only=production
 
# 2. esbuild でバンドル (tree-shaking 付き)
npx esbuild src/handler.ts \
  --bundle \
  --minify \
  --sourcemap \
  --platform=node \
  --target=node20 \
  --outfile=dist/handler.js \
  --external:@aws-sdk/*
 
# 3. ZIP パッケージの作成
cd dist
zip -r9 ../function.zip .
 
# AWS SDK v3 は Lambda ランタイムに組み込み済みのため
# --external:@aws-sdk/* で除外してサイズ削減

# メモリサイズの最適化を自動テストするスクリプト
import boto3
import json
import time
import statistics
 
lambda_client = boto3.client("lambda")
 
def benchmark_memory_sizes(function_name, payload, memory_sizes, iterations=10):
    """異なるメモリサイズでの実行時間を比較する"""
    results = {}
 
    for memory_size in memory_sizes:
        # メモリサイズを変更
        lambda_client.update_function_configuration(
            FunctionName=function_name,
            MemorySize=memory_size
        )
        time.sleep(5)  # 設定反映を待つ
 
        durations = []
        for i in range(iterations):
            response = lambda_client.invoke(
                FunctionName=function_name,
                Payload=json.dumps(payload)
            )
            # レスポンスヘッダから実行時間を取得
            log_result = response.get("LogResult", "")
            # Duration を解析
            duration = float(response["ResponseMetadata"]["HTTPHeaders"]
                           .get("x-amz-log-result", "0"))
            durations.append(duration)
 
        results[memory_size] = {
            "avg_duration_ms": statistics.mean(durations),
            "p99_duration_ms": sorted(durations)[int(len(durations) * 0.99)],
            "cost_per_invocation": (memory_size / 1024) * (statistics.mean(durations) / 1000) * 0.0000166667,
        }
 
    return results

# Provisioned Concurrency の設定
# まずエイリアスまたはバージョンを指定
aws lambda publish-version \
  --function-name my-api-function
 
aws lambda put-provisioned-concurrency-config \
  --function-name my-api-function \
  --qualifier 1 \
  --provisioned-concurrent-executions 50
 
# 状態確認
aws lambda get-provisioned-concurrency-config \
  --function-name my-api-function \
  --qualifier 1
 
# 設定一覧の確認
aws lambda list-provisioned-concurrency-configs \
  --function-name my-api-function
 
# Provisioned Concurrency の削除
aws lambda delete-provisioned-concurrency-config \
  --function-name my-api-function \
  --qualifier 1

# Auto Scaling ターゲットの登録
aws application-autoscaling register-scalable-target \
  --service-namespace lambda \
  --resource-id "function:my-api-function:prod" \
  --scalable-dimension "lambda:function:ProvisionedConcurrency" \
  --min-capacity 10 \
  --max-capacity 200
 
# ターゲット追跡スケーリングポリシー
aws application-autoscaling put-scaling-policy \
  --service-namespace lambda \
  --resource-id "function:my-api-function:prod" \
  --scalable-dimension "lambda:function:ProvisionedConcurrency" \
  --policy-name "provisioned-concurrency-target-tracking" \
  --policy-type TargetTrackingScaling \
  --target-tracking-scaling-policy-configuration '{
    "TargetValue": 0.7,
    "PredefinedMetricSpecification": {
      "PredefinedMetricType": "LambdaProvisionedConcurrencyUtilization"
    },
    "ScaleInCooldown": 300,
    "ScaleOutCooldown": 60
  }'
 
# スケジュールベースのスケーリング
aws application-autoscaling put-scheduled-action \
  --service-namespace lambda \
  --resource-id "function:my-api-function:prod" \
  --scalable-dimension "lambda:function:ProvisionedConcurrency" \
  --scheduled-action-name "morning-scale-up" \
  --schedule "cron(0 8 * * ? *)" \
  --scalable-target-action "MinCapacity=100,MaxCapacity=500"
 
# 夜間のスケールダウン
aws application-autoscaling put-scheduled-action \
  --service-namespace lambda \
  --resource-id "function:my-api-function:prod" \
  --scalable-dimension "lambda:function:ProvisionedConcurrency" \
  --scheduled-action-name "night-scale-down" \
  --schedule "cron(0 22 * * ? *)" \
  --scalable-target-action "MinCapacity=10,MaxCapacity=50"

# Reserved Concurrency の設定
aws lambda put-function-concurrency \
  --function-name my-api-function \
  --reserved-concurrent-executions 200
 
# Reserved Concurrency の確認
aws lambda get-function-concurrency \
  --function-name my-api-function
 
# Reserved Concurrency の削除 (アカウントプールに戻す)
aws lambda delete-function-concurrency \
  --function-name my-api-function

# Destinations の設定
aws lambda put-function-event-invoke-config \
  --function-name my-async-function \
  --maximum-retry-attempts 1 \
  --maximum-event-age-in-seconds 3600 \
  --destination-config '{
    "OnSuccess": {
      "Destination": "arn:aws:sqs:ap-northeast-1:123456789012:success-queue"
    },
    "OnFailure": {
      "Destination": "arn:aws:sqs:ap-northeast-1:123456789012:failure-queue"
    }
  }'
 
# 設定の確認
aws lambda get-function-event-invoke-config \
  --function-name my-async-function
 
# 設定の削除
aws lambda delete-function-event-invoke-config \
  --function-name my-async-function

{
  "version": "1.0",
  "timestamp": "2024-01-15T10:30:00.000Z",
  "requestContext": {
    "requestId": "abc123-def456-ghi789",
    "functionArn": "arn:aws:lambda:ap-northeast-1:123456789012:function:my-function",
    "condition": "Success",
    "approximateInvokeCount": 1
  },
  "requestPayload": {
    "orderId": "ORD-001",
    "amount": 5000
  },
  "responseContext": {
    "statusCode": 200,
    "executedVersion": "$LATEST",
    "functionError": null
  },
  "responsePayload": {
    "statusCode": 200,
    "body": "{\"message\": \"Order processed successfully\"}"
  }
}

# Lambda Destination を EventBridge に設定し、
# 複数の後続処理をイベントルールで分岐させるパターン
 
import json
import boto3
 
eventbridge = boto3.client("events")
 
def order_processor(event, context):
    """注文処理Lambda - 成功時にEventBridgeへ送信"""
    order_id = event["orderId"]
    amount = event["amount"]
 
    # 注文処理ロジック
    result = process_order(order_id, amount)
 
    return {
        "statusCode": 200,
        "orderId": order_id,
        "processedAmount": amount,
        "status": "COMPLETED"
    }
 
# EventBridge ルールでの後続処理分岐:
# ルール1: amount > 10000 → 高額注文通知 Lambda
# ルール2: 全注文 → 注文履歴 DynamoDB 書き込み Lambda
# ルール3: status=COMPLETED → 配送手配 Step Functions

# EventBridge ルール (CloudFormation)
HighValueOrderRule:
  Type: AWS::Events::Rule
  Properties:
    Name: high-value-order-notification
    EventPattern:
      source:
        - "lambda"
      detail-type:
        - "Lambda Function Invocation Result - Success"
      detail:
        requestContext:
          functionArn:
            - !GetAtt OrderProcessorFunction.Arn
        responsePayload:
          processedAmount:
            - numeric: [">=", 10000]
    Targets:
      - Arn: !GetAtt HighValueNotificationFunction.Arn
        Id: HighValueNotification

{
  "Comment": "注文処理ワークフロー",
  "StartAt": "ValidateInput",
  "States": {
    "ValidateInput": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:validate-input",
      "Next": "ProcessOrder",
      "Catch": [
        {
          "ErrorEquals": ["ValidationError"],
          "Next": "HandleError"
        }
      ]
    },
    "ProcessOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:process-order",
      "TimeoutSeconds": 300,
      "Retry": [
        {
          "ErrorEquals": ["States.TaskFailed"],
          "IntervalSeconds": 5,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ],
      "Next": "NotifySuccess",
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError"
        }
      ]
    },
    "NotifySuccess": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:notify-success",
      "End": true
    },
    "HandleError": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:handle-error",
      "Next": "NotifyFailure"
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:ap-northeast-1:123456789012:function:notify-failure",
      "End": true
    }
  }
}

{
  "Type": "Parallel",
  "Branches": [
    {
      "StartAt": "SendEmail",
      "States": {
        "SendEmail": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:...:send-email",
          "End": true
        }
      }
    },
    {
      "StartAt": "SendSMS",
      "States": {
        "SendSMS": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:...:send-sms",
          "End": true
        }
      }
    },
    {
      "StartAt": "UpdateDB",
      "States": {
        "UpdateDB": {
          "Type": "Task",
          "Resource": "arn:aws:lambda:...:update-db",
          "End": true
        }
      }
    }
  ],
  "Next": "AggregateResults"
}

{
  "Comment": "大量データの並列処理ワークフロー",
  "StartAt": "FetchItems",
  "States": {
    "FetchItems": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:fetch-items",
      "Next": "ProcessItems"
    },
    "ProcessItems": {
      "Type": "Map",
      "ItemsPath": "$.items",
      "MaxConcurrency": 10,
      "Iterator": {
        "StartAt": "ProcessSingleItem",
        "States": {
          "ProcessSingleItem": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:process-item",
            "Retry": [
              {
                "ErrorEquals": ["States.TaskFailed"],
                "IntervalSeconds": 2,
                "MaxAttempts": 3,
                "BackoffRate": 2.0
              }
            ],
            "End": true
          }
        }
      },
      "Next": "AggregateResults"
    },
    "AggregateResults": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:aggregate-results",
      "End": true
    }
  }
}

{
  "Comment": "S3 の大量ファイルを分散並列処理",
  "StartAt": "DistributedProcess",
  "States": {
    "DistributedProcess": {
      "Type": "Map",
      "ItemReader": {
        "Resource": "arn:aws:states:::s3:listObjectsV2",
        "Parameters": {
          "Bucket": "my-input-bucket",
          "Prefix": "data/"
        }
      },
      "ItemProcessor": {
        "ProcessorConfig": {
          "Mode": "DISTRIBUTED",
          "ExecutionType": "EXPRESS"
        },
        "StartAt": "ProcessFile",
        "States": {
          "ProcessFile": {
            "Type": "Task",
            "Resource": "arn:aws:lambda:...:process-file",
            "End": true
          }
        }
      },
      "MaxConcurrency": 1000,
      "Label": "DistributedProcess",
      "ResultWriter": {
        "Resource": "arn:aws:states:::s3:putObject",
        "Parameters": {
          "Bucket": "my-output-bucket",
          "Prefix": "results/"
        }
      },
      "End": true
    }
  }
}

{
  "Comment": "AWS SDK 統合による直接サービス呼び出し",
  "StartAt": "PutItemToDynamoDB",
  "States": {
    "PutItemToDynamoDB": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Parameters": {
        "TableName": "Orders",
        "Item": {
          "orderId": {"S.$": "$.orderId"},
          "status": {"S": "PENDING"},
          "createdAt": {"S.$": "$$.State.EnteredTime"}
        }
      },
      "Next": "PublishToSNS"
    },
    "PublishToSNS": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:ap-northeast-1:123456789012:order-notifications",
        "Message.$": "States.Format('New order: {}', $.orderId)",
        "Subject": "New Order Received"
      },
      "Next": "StartECSTask"
    },
    "StartECSTask": {
      "Type": "Task",
      "Resource": "arn:aws:states:::ecs:runTask.sync",
      "Parameters": {
        "LaunchType": "FARGATE",
        "Cluster": "arn:aws:ecs:...:cluster/my-cluster",
        "TaskDefinition": "arn:aws:ecs:...:task-definition/process-order:1",
        "Overrides": {
          "ContainerOverrides": [
            {
              "Name": "processor",
              "Environment": [
                {"Name": "ORDER_ID", "Value.$": "$.orderId"}
              ]
            }
          ]
        },
        "NetworkConfiguration": {
          "AwsvpcConfiguration": {
            "Subnets": ["subnet-111", "subnet-222"],
            "SecurityGroups": ["sg-12345678"]
          }
        }
      },
      "End": true
    }
  }
}

# SnapStart の有効化
aws lambda update-function-configuration \
  --function-name my-java-function \
  --snap-start '{"ApplyOn": "PublishedVersions"}'
 
# バージョン公開（スナップショット作成）
aws lambda publish-version \
  --function-name my-java-function
 
# SnapStart の状態確認
aws lambda get-function-configuration \
  --function-name my-java-function \
  --query 'SnapStart'

// SnapStart の afterRestore フック例
import org.crac.Context;
import org.crac.Core;
import org.crac.Resource;
 
public class MyHandler implements RequestHandler<APIGatewayProxyRequestEvent, APIGatewayProxyResponseEvent>,
                                   Resource {
 
    private Connection dbConnection;
 
    public MyHandler() {
        // Init Phase: CRaC リソースとして登録
        Core.getGlobalContext().register(this);
        // DB接続を確立
        this.dbConnection = DriverManager.getConnection(DB_URL);
    }
 
    @Override
    public void afterRestore(Context<? extends Resource> context) {
        // スナップショット復元後に呼ばれる
        // DB接続を再確立
        this.dbConnection = DriverManager.getConnection(DB_URL);
        // 乱数生成器を再シード
        SecureRandom.getInstanceStrong();
    }
 
    @Override
    public APIGatewayProxyResponseEvent handleRequest(
            APIGatewayProxyRequestEvent event, com.amazonaws.services.lambda.runtime.Context context) {
        // ハンドラロジック
        return new APIGatewayProxyResponseEvent().withStatusCode(200);
    }
}

# Python ライブラリのレイヤー作成
mkdir -p python/lib/python3.12/site-packages
pip install requests boto3-stubs[s3,dynamodb] \
  -t python/lib/python3.12/site-packages
 
# ZIP パッケージ作成
zip -r9 my-layer.zip python/
 
# レイヤーの公開
aws lambda publish-layer-version \
  --layer-name my-common-libs \
  --description "共通ライブラリ (requests, boto3-stubs)" \
  --compatible-runtimes python3.11 python3.12 \
  --compatible-architectures x86_64 arm64 \
  --zip-file fileb://my-layer.zip
 
# レイヤーを関数にアタッチ
aws lambda update-function-configuration \
  --function-name my-function \
  --layers \
    arn:aws:lambda:ap-northeast-1:123456789012:layer:my-common-libs:1 \
    arn:aws:lambda:ap-northeast-1:123456789012:layer:my-utilities:3
 
# レイヤーバージョンの一覧
aws lambda list-layer-versions \
  --layer-name my-common-libs
 
# レイヤーの削除
aws lambda delete-layer-version \
  --layer-name my-common-libs \
  --version-number 1

# レイヤーに含めるユーティリティモジュール
# python/lib/python3.12/site-packages/common/response.py
 
import json
from typing import Any, Optional
 
def api_response(
    status_code: int,
    body: Any,
    headers: Optional[dict] = None
) -> dict:
    """API Gateway 用の標準レスポンスを生成する"""
    default_headers = {
        "Content-Type": "application/json",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Headers": "Content-Type,Authorization",
        "Access-Control-Allow-Methods": "GET,POST,PUT,DELETE,OPTIONS",
    }
    if headers:
        default_headers.update(headers)
 
    return {
        "statusCode": status_code,
        "headers": default_headers,
        "body": json.dumps(body, ensure_ascii=False, default=str),
    }
 
def error_response(
    status_code: int,
    message: str,
    error_code: Optional[str] = None
) -> dict:
    """エラーレスポンスを生成する"""
    body = {"error": {"message": message}}
    if error_code:
        body["error"]["code"] = error_code
    return api_response(status_code, body)

# レイヤーに含めるロギングモジュール
# python/lib/python3.12/site-packages/common/logger.py
 
import json
import logging
import os
import sys
from datetime import datetime, timezone
 
class StructuredLogger:
    """構造化ログを出力するロガー"""
 
    def __init__(self, service_name: str = None):
        self.service_name = service_name or os.environ.get("SERVICE_NAME", "unknown")
        self.logger = logging.getLogger(self.service_name)
        self.logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))
 
        # JSON フォーマッタの設定
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(logging.Formatter("%(message)s"))
        self.logger.handlers = [handler]
 
    def _format(self, level: str, message: str, **kwargs) -> str:
        log_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": level,
            "service": self.service_name,
            "message": message,
            **kwargs
        }
        return json.dumps(log_entry, ensure_ascii=False, default=str)
 
    def info(self, message: str, **kwargs):
        self.logger.info(self._format("INFO", message, **kwargs))
 
    def error(self, message: str, **kwargs):
        self.logger.error(self._format("ERROR", message, **kwargs))
 
    def warning(self, message: str, **kwargs):
        self.logger.warning(self._format("WARNING", message, **kwargs))
 
    def debug(self, message: str, **kwargs):
        self.logger.debug(self._format("DEBUG", message, **kwargs))

# Lambda 関数の X-Ray トレーシングを有効化
aws lambda update-function-configuration \
  --function-name my-function \
  --tracing-config Mode=Active
 
# X-Ray トレースの取得
aws xray get-trace-summaries \
  --start-time $(date -d '1 hour ago' +%s) \
  --end-time $(date +%s) \
  --filter-expression 'service("my-function")'

# 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_data):
    """カスタムサブセグメントでビジネスロジックをトレース"""
 
    # アノテーションの追加 (フィルタリング用)
    subsegment = xray_recorder.current_subsegment()
    subsegment.put_annotation("order_id", order_data["orderId"])
    subsegment.put_annotation("customer_tier", order_data.get("tier", "standard"))
 
    # メタデータの追加 (デバッグ用)
    subsegment.put_metadata("order_details", order_data, "order")
 
    # ビジネスロジック
    result = validate_order(order_data)
    return result
 
def lambda_handler(event, context):
    return process_order(event)

# Lambda Insights の有効化 (拡張モニタリング)
aws lambda update-function-configuration \
  --function-name my-function \
  --layers \
    "arn:aws:lambda:ap-northeast-1:580247275435:layer:LambdaInsightsExtension:38"
 
# Lambda Insights が収集するメトリクス:
#   - cpu_total_time: CPU使用時間
#   - memory_utilization: メモリ使用率
#   - rx_bytes / tx_bytes: ネットワーク I/O
#   - init_duration: Init Phase の時間
#   - tmp_max: /tmp 使用量

# Embedded Metric Format (EMF) による
# Lambda からの高解像度カスタムメトリクス出力
 
import json
import time
 
def put_metric(namespace, metric_name, value, unit="None", dimensions=None):
    """EMF形式でCloudWatchカスタムメトリクスを出力"""
    emf_log = {
        "_aws": {
            "Timestamp": int(time.time() * 1000),
            "CloudWatchMetrics": [
                {
                    "Namespace": namespace,
                    "Dimensions": [list(dimensions.keys())] if dimensions else [[]],
                    "Metrics": [
                        {"Name": metric_name, "Unit": unit}
                    ]
                }
            ]
        },
        metric_name: value
    }
    if dimensions:
        emf_log.update(dimensions)
 
    # 標準出力に書くだけで CloudWatch メトリクスとして記録される
    print(json.dumps(emf_log))
 
def lambda_handler(event, context):
    start = time.time()
 
    # ビジネスロジック
    result = process_request(event)
 
    # カスタムメトリクスの出力
    elapsed = (time.time() - start) * 1000
    put_metric(
        "MyApplication",
        "ProcessingTime",
        elapsed,
        "Milliseconds",
        {"Environment": "production", "Service": "order-api"}
    )
    put_metric(
        "MyApplication",
        "OrdersProcessed",
        1,
        "Count",
        {"Environment": "production", "Service": "order-api"}
    )
 
    return result

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DynamoDBAccess",
      "Effect": "Allow",
      "Action": [
        "dynamodb:GetItem",
        "dynamodb:PutItem",
        "dynamodb:UpdateItem",
        "dynamodb:Query"
      ],
      "Resource": [
        "arn:aws:dynamodb:ap-northeast-1:123456789012:table/Orders",
        "arn:aws:dynamodb:ap-northeast-1:123456789012:table/Orders/index/*"
      ]
    },
    {
      "Sid": "S3ReadAccess",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::my-config-bucket/config/*"
    },
    {
      "Sid": "SecretsManagerAccess",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": "arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:my-api-key-*"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:ap-northeast-1:123456789012:log-group:/aws/lambda/my-function:*"
    }
  ]
}

# Secrets Manager からシークレットを取得する
# Lambda Extensions を使ったキャッシュ方式
 
import json
import os
import urllib3
 
# Lambda Extensions のキャッシュポート
SECRETS_EXTENSION_PORT = 2773
http = urllib3.PoolManager()
 
def get_secret(secret_name):
    """Secrets Manager Lambda Extension 経由でシークレットを取得"""
    url = (
        f"http://localhost:{SECRETS_EXTENSION_PORT}"
        f"/secretsmanager/get?secretId={secret_name}"
    )
    headers = {
        "X-Aws-Parameters-Secrets-Token": os.environ["AWS_SESSION_TOKEN"]
    }
    response = http.request("GET", url, headers=headers)
    return json.loads(response.data)["SecretString"]
 
# Init Phase でシークレットを取得 (キャッシュされる)
DB_CREDENTIALS = json.loads(get_secret("prod/db-credentials"))
API_KEY = get_secret("prod/external-api-key")
 
def lambda_handler(event, context):
    # シークレットを使用
    db_host = DB_CREDENTIALS["host"]
    db_password = DB_CREDENTIALS["password"]
    # ...

# VPC Lambda 用のセキュリティグループ作成
aws ec2 create-security-group \
  --group-name lambda-sg \
  --description "Security group for Lambda functions" \
  --vpc-id vpc-12345678
 
# RDS へのアクセスを許可 (RDS の SG に追加)
aws ec2 authorize-security-group-ingress \
  --group-id sg-rds-12345678 \
  --protocol tcp \
  --port 3306 \
  --source-group sg-lambda-12345678
 
# VPC Endpoint の作成 (DynamoDB)
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-12345678 \
  --service-name com.amazonaws.ap-northeast-1.dynamodb \
  --route-table-ids rtb-12345678
 
# VPC Endpoint の作成 (S3)
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-12345678 \
  --service-name com.amazonaws.ap-northeast-1.s3 \
  --route-table-ids rtb-12345678
 
# VPC Endpoint の作成 (SQS - Interface 型)
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-12345678 \
  --service-name com.amazonaws.ap-northeast-1.sqs \
  --vpc-endpoint-type Interface \
  --subnet-ids subnet-111 subnet-222 \
  --security-group-ids sg-vpce-12345678

# Lambda コンテナイメージの Dockerfile 例 (Python)
FROM public.ecr.aws/lambda/python:3.12
 
# 依存関係のインストール
COPY requirements.txt ${LAMBDA_TASK_ROOT}/
RUN pip install -r ${LAMBDA_TASK_ROOT}/requirements.txt --no-cache-dir
 
# 関数コードのコピー
COPY app/ ${LAMBDA_TASK_ROOT}/app/
COPY lambda_function.py ${LAMBDA_TASK_ROOT}/
 
# ハンドラの指定
CMD ["lambda_function.lambda_handler"]

# コンテナイメージのビルドとデプロイ
# 1. ECR リポジトリの作成
aws ecr create-repository \
  --repository-name my-lambda-function \
  --image-scanning-configuration scanOnPush=true
 
# 2. イメージのビルド
docker build -t my-lambda-function:latest \
  --platform linux/amd64 .
 
# 3. ECR にプッシュ
ECR_URI=123456789012.dkr.ecr.ap-northeast-1.amazonaws.com
aws ecr get-login-password | docker login --username AWS --password-stdin ${ECR_URI}
docker tag my-lambda-function:latest ${ECR_URI}/my-lambda-function:latest
docker push ${ECR_URI}/my-lambda-function:latest
 
# 4. Lambda 関数の作成
aws lambda create-function \
  --function-name my-container-function \
  --package-type Image \
  --code ImageUri=${ECR_URI}/my-lambda-function:latest \
  --role arn:aws:iam::123456789012:role/lambda-execution-role \
  --memory-size 1024 \
  --timeout 30

# マルチステージビルドで軽量なLambdaコンテナを作成
FROM python:3.12-slim AS builder
 
WORKDIR /build
COPY requirements.txt .
RUN pip install --user -r requirements.txt --no-cache-dir
 
# 本番ステージ
FROM public.ecr.aws/lambda/python:3.12
 
# ビルドステージからの依存関係コピー
COPY --from=builder /root/.local/lib/python3.12/site-packages ${LAMBDA_TASK_ROOT}/
COPY app/ ${LAMBDA_TASK_ROOT}/app/
COPY lambda_function.py ${LAMBDA_TASK_ROOT}/
 
CMD ["lambda_function.lambda_handler"]

# 関数 URL の作成 (IAM 認証なし)
aws lambda create-function-url-config \
  --function-name my-api-function \
  --auth-type NONE \
  --cors '{
    "AllowCredentials": false,
    "AllowHeaders": ["content-type", "authorization"],
    "AllowMethods": ["GET", "POST", "PUT", "DELETE"],
    "AllowOrigins": ["https://example.com"],
    "ExposeHeaders": ["x-request-id"],
    "MaxAge": 86400
  }'
 
# リソースベースポリシーの追加 (パブリックアクセス)
aws lambda add-permission \
  --function-name my-api-function \
  --statement-id AllowPublicAccess \
  --action lambda:InvokeFunctionUrl \
  --principal "*" \
  --function-url-auth-type NONE
 
# 関数 URL の取得
aws lambda get-function-url-config \
  --function-name my-api-function
 
# IAM 認証付き関数 URL
aws lambda create-function-url-config \
  --function-name my-internal-api \
  --auth-type AWS_IAM

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: 'Lambda 応用構成テンプレート'
 
Parameters:
  EnvironmentName:
    Type: String
    Default: dev
    AllowedValues: [dev, stg, prod]
 
  ProvisionedConcurrency:
    Type: Number
    Default: 0
    Description: 'Provisioned Concurrency 数 (0=無効)'
 
Conditions:
  EnableProvisionedConcurrency: !Not [!Equals [!Ref ProvisionedConcurrency, 0]]
  IsProduction: !Equals [!Ref EnvironmentName, prod]
 
Globals:
  Function:
    Runtime: python3.12
    MemorySize: 1024
    Timeout: 30
    Tracing: Active
    Environment:
      Variables:
        ENVIRONMENT: !Ref EnvironmentName
        TABLE_NAME: !Ref OrdersTable
        LOG_LEVEL: !If [IsProduction, INFO, DEBUG]
 
Resources:
  # Lambda 関数
  OrderApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: !Sub '${EnvironmentName}-order-api'
      Handler: app.lambda_handler
      CodeUri: src/order-api/
      AutoPublishAlias: live
      ProvisionedConcurrencyConfig:
        !If
        - EnableProvisionedConcurrency
        - ProvisionedConcurrentExecutions: !Ref ProvisionedConcurrency
        - !Ref AWS::NoValue
      DeploymentPreference:
        Type: !If [IsProduction, Linear10PercentEvery1Minute, AllAtOnce]
        Alarms:
          - !Ref OrderApiErrorAlarm
      Events:
        GetOrder:
          Type: Api
          Properties:
            Path: /orders/{orderId}
            Method: get
            RestApiId: !Ref ApiGateway
        CreateOrder:
          Type: Api
          Properties:
            Path: /orders
            Method: post
            RestApiId: !Ref ApiGateway
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref OrdersTable
        - SQSSendMessagePolicy:
            QueueName: !GetAtt OrderQueue.QueueName
 
  # Step Functions ステートマシン
  OrderProcessingStateMachine:
    Type: AWS::StepFunctions::StateMachine
    Properties:
      StateMachineName: !Sub '${EnvironmentName}-order-processing'
      DefinitionString: !Sub |
        {
          "Comment": "注文処理ワークフロー",
          "StartAt": "ValidateOrder",
          "States": {
            "ValidateOrder": {
              "Type": "Task",
              "Resource": "${ValidateOrderFunction.Arn}",
              "Next": "ProcessPayment",
              "Catch": [{"ErrorEquals": ["ValidationError"], "Next": "FailOrder"}]
            },
            "ProcessPayment": {
              "Type": "Task",
              "Resource": "${ProcessPaymentFunction.Arn}",
              "Retry": [{"ErrorEquals": ["States.TaskFailed"], "IntervalSeconds": 5, "MaxAttempts": 3, "BackoffRate": 2.0}],
              "Next": "NotifySuccess",
              "Catch": [{"ErrorEquals": ["States.ALL"], "Next": "FailOrder"}]
            },
            "NotifySuccess": {
              "Type": "Task",
              "Resource": "arn:aws:states:::sns:publish",
              "Parameters": {
                "TopicArn": "${OrderNotificationTopic}",
                "Message.$": "States.Format('Order {} processed successfully', $.orderId)"
              },
              "End": true
            },
            "FailOrder": {
              "Type": "Task",
              "Resource": "${FailOrderFunction.Arn}",
              "End": true
            }
          }
        }
      RoleArn: !GetAtt StepFunctionsRole.Arn
 
  # DynamoDB テーブル
  OrdersTable:
    Type: AWS::DynamoDB::Table
    DeletionPolicy: Retain
    Properties:
      TableName: !Sub '${EnvironmentName}-orders'
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: orderId
          AttributeType: S
        - AttributeName: customerId
          AttributeType: S
      KeySchema:
        - AttributeName: orderId
          KeyType: HASH
      GlobalSecondaryIndexes:
        - IndexName: customer-index
          KeySchema:
            - AttributeName: customerId
              KeyType: HASH
          Projection:
            ProjectionType: ALL
 
  # CloudWatch アラーム
  OrderApiErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub '${EnvironmentName}-order-api-errors'
      MetricName: Errors
      Namespace: AWS/Lambda
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 3
      Threshold: 5
      ComparisonOperator: GreaterThanOrEqualToThreshold
      Dimensions:
        - Name: FunctionName
          Value: !Ref OrderApiFunction
      AlarmActions:
        - !Ref AlertTopic
 
Outputs:
  ApiEndpoint:
    Description: API Gateway エンドポイント
    Value: !Sub 'https://${ApiGateway}.execute-api.${AWS::Region}.amazonaws.com/Prod'
 
  StateMachineArn:
    Description: Step Functions ステートマシン ARN
    Value: !Ref OrderProcessingStateMachine