# handler.py -- CRUD API のエントリポイント
import json
import boto3
import os
from decimal import Decimal
 
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
 
class DecimalEncoder(json.JSONEncoder):
    """DynamoDB の Decimal 型を JSON シリアライズするヘルパー"""
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return super().default(obj)
 
def lambda_handler(event, context):
    http_method = event["httpMethod"]
    path = event.get("pathParameters") or {}
 
    try:
        if http_method == "GET" and "id" in path:
            return get_item(path["id"])
        elif http_method == "GET":
            return list_items(event)
        elif http_method == "POST":
            return create_item(json.loads(event["body"]))
        elif http_method == "PUT" and "id" in path:
            return update_item(path["id"], json.loads(event["body"]))
        elif http_method == "DELETE" and "id" in path:
            return delete_item(path["id"])
        else:
            return response(404, {"error": "Not found"})
    except Exception as e:
        return response(500, {"error": str(e)})
 
def get_item(item_id):
    result = table.get_item(Key={"id": item_id})
    if "Item" in result:
        return response(200, result["Item"])
    return response(404, {"error": "Item not found"})
 
def list_items(event):
    params = event.get("queryStringParameters") or {}
    limit = int(params.get("limit", "20"))
    result = table.scan(Limit=limit)
    return response(200, {"items": result["Items"], "count": result["Count"]})
 
def create_item(body):
    import uuid
    body["id"] = str(uuid.uuid4())
    table.put_item(Item=body)
    return response(201, body)
 
def update_item(item_id, body):
    body["id"] = item_id
    table.put_item(Item=body)
    return response(200, body)
 
def delete_item(item_id):
    table.delete_item(Key={"id": item_id})
    return response(204, None)
 
def response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        },
        "body": json.dumps(body, cls=DecimalEncoder) if body else ""
    }

# template.yaml (AWS SAM)
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Serverless CRUD API
 
Globals:
  Function:
    Runtime: python3.12
    Timeout: 30
    MemorySize: 256
    Environment:
      Variables:
        TABLE_NAME: !Ref ItemsTable
    Tracing: Active
    Layers:
      - !Ref SharedLayer
 
Resources:
  # Lambda レイヤー（共通ライブラリ）
  SharedLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      LayerName: shared-utils
      ContentUri: layers/shared/
      CompatibleRuntimes:
        - python3.12
 
  # API 関数
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: handler.lambda_handler
      CodeUri: src/
      Events:
        GetItem:
          Type: Api
          Properties:
            Path: /items/{id}
            Method: get
        ListItems:
          Type: Api
          Properties:
            Path: /items
            Method: get
        CreateItem:
          Type: Api
          Properties:
            Path: /items
            Method: post
        UpdateItem:
          Type: Api
          Properties:
            Path: /items/{id}
            Method: put
        DeleteItem:
          Type: Api
          Properties:
            Path: /items/{id}
            Method: delete
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref ItemsTable
 
  ItemsTable:
    Type: AWS::DynamoDB::Table
    Properties:
      TableName: items
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      PointInTimeRecoverySpecification:
        PointInTimeRecoveryEnabled: true
      SSESpecification:
        SSEEnabled: true

# handler_with_powertools.py -- Lambda Powertools を活用した実装
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.utilities.typing import LambdaContext
import boto3
import os
 
logger = Logger(service="items-api")
tracer = Tracer(service="items-api")
metrics = Metrics(namespace="ItemsAPI", service="items-api")
app = APIGatewayRestResolver()
 
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
 
@app.get("/items/<item_id>")
@tracer.capture_method
def get_item(item_id: str):
    logger.info("Getting item", extra={"item_id": item_id})
    result = table.get_item(Key={"id": item_id})
    if "Item" not in result:
        raise app.not_found()
    metrics.add_metric(name="ItemRetrieved", unit=MetricUnit.Count, value=1)
    return result["Item"]
 
@app.get("/items")
@tracer.capture_method
def list_items():
    params = app.current_event.query_string_parameters or {}
    limit = int(params.get("limit", "20"))
    logger.info("Listing items", extra={"limit": limit})
    result = table.scan(Limit=limit)
    metrics.add_metric(name="ItemsListed", unit=MetricUnit.Count, value=result["Count"])
    return {"items": result["Items"], "count": result["Count"]}
 
@app.post("/items")
@tracer.capture_method
def create_item():
    import uuid
    body = app.current_event.json_body
    body["id"] = str(uuid.uuid4())
    table.put_item(Item=body)
    logger.info("Item created", extra={"item_id": body["id"]})
    metrics.add_metric(name="ItemCreated", unit=MetricUnit.Count, value=1)
    return body, 201
 
@app.put("/items/<item_id>")
@tracer.capture_method
def update_item(item_id: str):
    body = app.current_event.json_body
    body["id"] = item_id
    table.put_item(Item=body)
    logger.info("Item updated", extra={"item_id": item_id})
    metrics.add_metric(name="ItemUpdated", unit=MetricUnit.Count, value=1)
    return body
 
@app.delete("/items/<item_id>")
@tracer.capture_method
def delete_item(item_id: str):
    table.delete_item(Key={"id": item_id})
    logger.info("Item deleted", extra={"item_id": item_id})
    metrics.add_metric(name="ItemDeleted", unit=MetricUnit.Count, value=1)
    return "", 204
 
@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

# AppSync GraphQL API の SAM テンプレート
Resources:
  GraphQLApi:
    Type: AWS::AppSync::GraphQLApi
    Properties:
      Name: items-graphql-api
      AuthenticationType: AMAZON_COGNITO_USER_POOLS
      UserPoolConfig:
        UserPoolId: !Ref UserPool
        DefaultAction: ALLOW
        AwsRegion: !Ref AWS::Region
      LogConfig:
        CloudWatchLogsRoleArn: !GetAtt AppSyncLogsRole.Arn
        FieldLogLevel: ERROR
      XrayEnabled: true
 
  GraphQLSchema:
    Type: AWS::AppSync::GraphQLSchema
    Properties:
      ApiId: !GetAtt GraphQLApi.ApiId
      Definition: |
        type Item {
          id: ID!
          name: String!
          description: String
          price: Float
          category: String
          createdAt: AWSDateTime
          updatedAt: AWSDateTime
        }
 
        input CreateItemInput {
          name: String!
          description: String
          price: Float
          category: String
        }
 
        input UpdateItemInput {
          name: String
          description: String
          price: Float
          category: String
        }
 
        type ItemConnection {
          items: [Item]
          nextToken: String
        }
 
        type Query {
          getItem(id: ID!): Item
          listItems(limit: Int, nextToken: String): ItemConnection
          searchItems(keyword: String!, limit: Int): ItemConnection
        }
 
        type Mutation {
          createItem(input: CreateItemInput!): Item
          updateItem(id: ID!, input: UpdateItemInput!): Item
          deleteItem(id: ID!): Item
        }
 
        type Subscription {
          onCreateItem: Item
            @aws_subscribe(mutations: ["createItem"])
          onUpdateItem(id: ID): Item
            @aws_subscribe(mutations: ["updateItem"])
        }
 
        schema {
          query: Query
          mutation: Mutation
          subscription: Subscription
        }

# image_processor.py -- S3 トリガーによる画像リサイズ
import boto3
import os
from PIL import Image
import io
 
s3 = boto3.client("s3")
DEST_BUCKET = os.environ["DEST_BUCKET"]
SIZES = [(128, 128), (256, 256), (512, 512)]
 
def lambda_handler(event, context):
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]
 
        # 元画像のダウンロード
        response = s3.get_object(Bucket=bucket, Key=key)
        image = Image.open(io.BytesIO(response["Body"].read()))
 
        # 各サイズにリサイズしてアップロード
        for width, height in SIZES:
            resized = image.copy()
            resized.thumbnail((width, height))
 
            buffer = io.BytesIO()
            resized.save(buffer, format="JPEG", quality=85)
            buffer.seek(0)
 
            dest_key = f"thumbnails/{width}x{height}/{key}"
            s3.put_object(
                Bucket=DEST_BUCKET,
                Key=dest_key,
                Body=buffer,
                ContentType="image/jpeg"
            )
 
    return {"processed": len(event["Records"])}

{
  "Source": "com.myapp.orders",
  "DetailType": "OrderCreated",
  "Detail": {
    "orderId": "ORD-12345",
    "customerId": "CUST-67890",
    "amount": 15000,
    "items": [
      {"productId": "PROD-001", "quantity": 2}
    ]
  }
}

# EventBridge ルールの作成
aws events put-rule \
  --name "high-value-orders" \
  --event-pattern '{
    "source": ["com.myapp.orders"],
    "detail-type": ["OrderCreated"],
    "detail": {
      "amount": [{"numeric": [">=", 10000]}]
    }
  }'
 
# ターゲットの設定（Lambda 関数）
aws events put-targets \
  --rule "high-value-orders" \
  --targets '[
    {
      "Id": "notify-vip-team",
      "Arn": "arn:aws:lambda:...:notify-vip-team"
    },
    {
      "Id": "fraud-check",
      "Arn": "arn:aws:lambda:...:fraud-check"
    }
  ]'

# EventBridge Pipes の SAM テンプレート
Resources:
  OrderProcessingPipe:
    Type: AWS::Pipes::Pipe
    Properties:
      Name: order-processing-pipe
      RoleArn: !GetAtt PipeRole.Arn
      Source: !GetAtt OrderQueue.Arn
      SourceParameters:
        SqsQueueParameters:
          BatchSize: 10
          MaximumBatchingWindowInSeconds: 30
        FilterCriteria:
          Filters:
            - Pattern: '{"body": {"orderType": ["premium"]}}'
      Enrichment: !GetAtt EnrichmentFunction.Arn
      Target: !Ref OrderProcessingStateMachine
      TargetParameters:
        StepFunctionStateMachineParameters:
          InvocationType: FIRE_AND_FORGET

# dynamodb_stream_handler.py -- DynamoDB Streams CDC パターン
import json
import boto3
import os
from datetime import datetime
 
sqs = boto3.client("sqs")
AUDIT_QUEUE_URL = os.environ["AUDIT_QUEUE_URL"]
 
def lambda_handler(event, context):
    """DynamoDB Streams からのイベントを処理し、監査キューに送信"""
    for record in event["Records"]:
        event_name = record["eventName"]  # INSERT, MODIFY, REMOVE
        event_id = record["eventID"]
        timestamp = record["dynamodb"]["ApproximateCreationDateTime"]
 
        audit_event = {
            "eventId": event_id,
            "eventType": event_name,
            "timestamp": str(timestamp),
            "tableName": record["eventSourceARN"].split("/")[1],
            "processedAt": datetime.utcnow().isoformat()
        }
 
        if event_name in ("INSERT", "MODIFY"):
            new_image = deserialize(record["dynamodb"]["NewImage"])
            audit_event["newImage"] = new_image
 
        if event_name in ("MODIFY", "REMOVE"):
            old_image = deserialize(record["dynamodb"]["OldImage"])
            audit_event["oldImage"] = old_image
 
        if event_name == "MODIFY":
            # 変更フィールドの検出
            changes = detect_changes(
                deserialize(record["dynamodb"]["OldImage"]),
                deserialize(record["dynamodb"]["NewImage"])
            )
            audit_event["changedFields"] = changes
 
        sqs.send_message(
            QueueUrl=AUDIT_QUEUE_URL,
            MessageBody=json.dumps(audit_event),
            MessageGroupId=audit_event.get("newImage", {}).get("id", "default")
        )
 
    return {"batchItemFailures": []}
 
def deserialize(image):
    """DynamoDB の型付き形式を通常の dict に変換"""
    from boto3.dynamodb.types import TypeDeserializer
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in image.items()}
 
def detect_changes(old_image, new_image):
    """新旧イメージの差分を検出"""
    changes = []
    all_keys = set(list(old_image.keys()) + list(new_image.keys()))
    for key in all_keys:
        old_val = old_image.get(key)
        new_val = new_image.get(key)
        if old_val != new_val:
            changes.append({
                "field": key,
                "oldValue": str(old_val),
                "newValue": str(new_val)
            })
    return changes

# order_publisher.py -- SNS へのイベント発行
import boto3
import json
import os
 
sns = boto3.client("sns")
TOPIC_ARN = os.environ["ORDER_TOPIC_ARN"]
 
def lambda_handler(event, context):
    order = json.loads(event["body"])
 
    # SNS トピックにパブリッシュ
    sns.publish(
        TopicArn=TOPIC_ARN,
        Message=json.dumps(order),
        MessageAttributes={
            "orderType": {
                "DataType": "String",
                "StringValue": order.get("type", "standard")
            },
            "amount": {
                "DataType": "Number",
                "StringValue": str(order["amount"])
            }
        }
    )
 
    return {
        "statusCode": 202,
        "body": json.dumps({"message": "Order accepted", "orderId": order["id"]})
    }

# batch_processor.py -- SQS バッチ処理 with 部分失敗報告
import json
 
def lambda_handler(event, context):
    batch_item_failures = []
 
    for record in event["Records"]:
        try:
            body = json.loads(record["body"])
            # SNS でラップされている場合
            if "Message" in body:
                message = json.loads(body["Message"])
            else:
                message = body
 
            process_order(message)
        except Exception as e:
            print(f"Error processing {record['messageId']}: {e}")
            batch_item_failures.append({
                "itemIdentifier": record["messageId"]
            })
 
    return {"batchItemFailures": batch_item_failures}
 
def process_order(order):
    # 注文処理ロジック
    print(f"Processing order: {order['id']}")

{
  "orderType": ["premium", "vip"],
  "amount": [{"numeric": [">=", 10000]}],
  "region": [{"prefix": "asia-"}]
}

# SAM テンプレートでの SNS + SQS ファンアウト定義
Resources:
  OrderTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: order-events
 
  # 在庫更新キュー - 全注文を受信
  InventoryQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: inventory-updates
      VisibilityTimeout: 60
      RedrivePolicy:
        deadLetterTargetArn: !GetAtt InventoryDLQ.Arn
        maxReceiveCount: 3
 
  InventoryDLQ:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: inventory-updates-dlq
      MessageRetentionPeriod: 1209600  # 14日
 
  InventorySubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref OrderTopic
      Protocol: sqs
      Endpoint: !GetAtt InventoryQueue.Arn
      RawMessageDelivery: true
 
  # VIP 通知キュー - 高額注文のみ受信
  VipNotificationQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: vip-notifications
      VisibilityTimeout: 30
 
  VipSubscription:
    Type: AWS::SNS::Subscription
    Properties:
      TopicArn: !Ref OrderTopic
      Protocol: sqs
      Endpoint: !GetAtt VipNotificationQueue.Arn
      RawMessageDelivery: true
      FilterPolicy:
        orderType:
          - premium
          - vip
        amount:
          - numeric:
              - ">="
              - 10000
 
  # SQS ポリシー（SNS からの送信許可）
  InventoryQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues:
        - !Ref InventoryQueue
      PolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              Service: sns.amazonaws.com
            Action: sqs:SendMessage
            Resource: !GetAtt InventoryQueue.Arn
            Condition:
              ArnEquals:
                aws:SourceArn: !Ref OrderTopic

# stream_processor.py -- DynamoDB Streams から読取りモデルを更新
import boto3
import os
import json
 
opensearch_endpoint = os.environ["OPENSEARCH_ENDPOINT"]
 
def lambda_handler(event, context):
    for record in event["Records"]:
        event_name = record["eventName"]  # INSERT, MODIFY, REMOVE
 
        if event_name in ("INSERT", "MODIFY"):
            new_image = record["dynamodb"]["NewImage"]
            item = deserialize_dynamodb(new_image)
            index_to_opensearch(item)
        elif event_name == "REMOVE":
            old_image = record["dynamodb"]["OldImage"]
            item = deserialize_dynamodb(old_image)
            remove_from_opensearch(item["id"])
 
def deserialize_dynamodb(image):
    """DynamoDB の型付き形式を通常の dict に変換"""
    from boto3.dynamodb.types import TypeDeserializer
    deserializer = TypeDeserializer()
    return {k: deserializer.deserialize(v) for k, v in image.items()}
 
def index_to_opensearch(item):
    """OpenSearch にドキュメントをインデックス"""
    import requests
    from requests_aws4auth import AWS4Auth
 
    credentials = boto3.Session().get_credentials()
    auth = AWS4Auth(
        credentials.access_key, credentials.secret_key,
        os.environ["AWS_REGION"], "es",
        session_token=credentials.token
    )
 
    url = f"{opensearch_endpoint}/items/_doc/{item['id']}"
    requests.put(url, auth=auth, json=item)

# cache_updater.py -- DynamoDB Streams から ElastiCache (Redis) を更新
import boto3
import json
import os
import redis
 
redis_client = redis.Redis(
    host=os.environ["REDIS_ENDPOINT"],
    port=6379,
    decode_responses=True,
    ssl=True
)
 
def lambda_handler(event, context):
    for record in event["Records"]:
        event_name = record["eventName"]
 
        if event_name in ("INSERT", "MODIFY"):
            new_image = deserialize(record["dynamodb"]["NewImage"])
            item_id = new_image["id"]
 
            # 個別アイテムのキャッシュ更新
            redis_client.set(
                f"item:{item_id}",
                json.dumps(new_image),
                ex=3600  # 1時間TTL
            )
 
            # カテゴリ別ソート済みセットの更新
            if "category" in new_image and "updatedAt" in new_image:
                redis_client.zadd(
                    f"category:{new_image['category']}",
                    {item_id: float(new_image["updatedAt"])}
                )
 
            # 検索用の逆引きインデックス更新
            if "tags" in new_image:
                for tag in new_image["tags"]:
                    redis_client.sadd(f"tag:{tag}", item_id)
 
        elif event_name == "REMOVE":
            old_image = deserialize(record["dynamodb"]["OldImage"])
            item_id = old_image["id"]
 
            redis_client.delete(f"item:{item_id}")
 
            if "category" in old_image:
                redis_client.zrem(f"category:{old_image['category']}", item_id)
 
            if "tags" in old_image:
                for tag in old_image["tags"]:
                    redis_client.srem(f"tag:{tag}", item_id)
 
    return {"batchItemFailures": []}
 
def deserialize(image):
    from boto3.dynamodb.types import TypeDeserializer
    d = TypeDeserializer()
    return {k: d.deserialize(v) for k, v in image.items()}

{
  "Comment": "Order Saga",
  "StartAt": "ReserveInventory",
  "States": {
    "ReserveInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:reserve-inventory",
      "Next": "ProcessPayment",
      "Catch": [{"ErrorEquals": ["States.ALL"], "Next": "Fail"}]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:process-payment",
      "Next": "ArrangeShipping",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "CancelReservation"
      }]
    },
    "ArrangeShipping": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:arrange-shipping",
      "Next": "ConfirmOrder",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "RefundPayment"
      }]
    },
    "ConfirmOrder": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:confirm-order",
      "End": true
    },
    "RefundPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:refund-payment",
      "Next": "CancelReservation"
    },
    "CancelReservation": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:cancel-reservation",
      "Next": "Fail"
    },
    "Fail": {
      "Type": "Fail",
      "Error": "OrderFailed",
      "Cause": "Saga compensation completed"
    }
  }
}

# Step Functions Express Workflow の SAM テンプレート
Resources:
  OrderProcessingStateMachine:
    Type: AWS::Serverless::StateMachine
    Properties:
      DefinitionUri: statemachine/order-processing.asl.json
      Type: EXPRESS
      Tracing:
        Enabled: true
      Logging:
        Destinations:
          - CloudWatchLogsLogGroup:
              LogGroupArn: !GetAtt StateMachineLogGroup.Arn
        IncludeExecutionData: true
        Level: ALL
      Policies:
        - LambdaInvokePolicy:
            FunctionName: !Ref ReserveInventoryFunction
        - LambdaInvokePolicy:
            FunctionName: !Ref ProcessPaymentFunction
        - LambdaInvokePolicy:
            FunctionName: !Ref ArrangeShippingFunction
 
  StateMachineLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: /aws/stepfunctions/order-processing
      RetentionInDays: 30

{
  "Comment": "Parallel processing with error handling",
  "StartAt": "ValidateInput",
  "States": {
    "ValidateInput": {
      "Type": "Pass",
      "Next": "ParallelProcessing"
    },
    "ParallelProcessing": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "ProcessImages",
          "States": {
            "ProcessImages": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:process-images",
              "Retry": [
                {
                  "ErrorEquals": ["ServiceException", "TooManyRequestsException"],
                  "IntervalSeconds": 2,
                  "MaxAttempts": 3,
                  "BackoffRate": 2.0
                }
              ],
              "End": true
            }
          }
        },
        {
          "StartAt": "ProcessMetadata",
          "States": {
            "ProcessMetadata": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:process-metadata",
              "Retry": [
                {
                  "ErrorEquals": ["States.TaskFailed"],
                  "IntervalSeconds": 1,
                  "MaxAttempts": 2,
                  "BackoffRate": 2.0
                }
              ],
              "End": true
            }
          }
        },
        {
          "StartAt": "SendNotification",
          "States": {
            "SendNotification": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:...:send-notification",
              "End": true
            }
          }
        }
      ],
      "Catch": [
        {
          "ErrorEquals": ["States.ALL"],
          "Next": "HandleError",
          "ResultPath": "$.error"
        }
      ],
      "Next": "AggregateResults"
    },
    "AggregateResults": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:aggregate-results",
      "End": true
    },
    "HandleError": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:handle-error",
      "End": true
    }
  }
}

# kinesis_processor.py -- Kinesis Data Streams のレコード処理
import json
import base64
import boto3
import os
from datetime import datetime
from collections import defaultdict
 
dynamodb = boto3.resource("dynamodb")
metrics_table = dynamodb.Table(os.environ["METRICS_TABLE"])
 
def lambda_handler(event, context):
    """Kinesis ストリームからのイベントを集約処理"""
    batch_item_failures = []
    aggregated = defaultdict(lambda: {"count": 0, "total_value": 0})
 
    for record in event["Records"]:
        try:
            # Kinesis レコードのデコード
            payload = base64.b64decode(record["kinesis"]["data"]).decode("utf-8")
            data = json.loads(payload)
 
            # 時間ウィンドウでの集約
            timestamp = datetime.fromisoformat(data["timestamp"])
            window_key = timestamp.strftime("%Y-%m-%dT%H:%M")  # 分単位
            metric_key = f"{data['metric_name']}#{window_key}"
 
            aggregated[metric_key]["count"] += 1
            aggregated[metric_key]["total_value"] += data.get("value", 0)
            aggregated[metric_key]["metric_name"] = data["metric_name"]
            aggregated[metric_key]["window"] = window_key
 
        except Exception as e:
            print(f"Error processing record {record['kinesis']['sequenceNumber']}: {e}")
            batch_item_failures.append({
                "itemIdentifier": record["kinesis"]["sequenceNumber"]
            })
 
    # 集約結果をDynamoDBに書き込み
    with metrics_table.batch_writer() as batch:
        for key, agg in aggregated.items():
            batch.put_item(Item={
                "pk": key,
                "metric_name": agg["metric_name"],
                "window": agg["window"],
                "count": agg["count"],
                "total_value": int(agg["total_value"]),
                "avg_value": int(agg["total_value"] / agg["count"]),
                "ttl": int(datetime.utcnow().timestamp()) + 86400 * 7  # 7日保持
            })
 
    return {"batchItemFailures": batch_item_failures}

# Kinesis Data Streams + Lambda の SAM テンプレート
Resources:
  ClickStream:
    Type: AWS::Kinesis::Stream
    Properties:
      Name: click-stream
      ShardCount: 4
      StreamModeDetails:
        StreamMode: ON_DEMAND  # 自動スケーリング
      RetentionPeriodHours: 168  # 7日間保持
      StreamEncryption:
        EncryptionType: KMS
        KeyId: alias/aws/kinesis
 
  StreamProcessor:
    Type: AWS::Serverless::Function
    Properties:
      Handler: kinesis_processor.lambda_handler
      Runtime: python3.12
      MemorySize: 512
      Timeout: 300
      Events:
        KinesisEvent:
          Type: Kinesis
          Properties:
            Stream: !GetAtt ClickStream.Arn
            StartingPosition: LATEST
            BatchSize: 100
            MaximumBatchingWindowInSeconds: 30
            ParallelizationFactor: 10
            MaximumRetryAttempts: 3
            BisectBatchOnFunctionError: true
            DestinationConfig:
              OnFailure:
                Destination: !GetAtt FailedRecordsDLQ.Arn
            FunctionResponseTypes:
              - ReportBatchItemFailures

# scheduled_report.py -- 日次レポート生成
import boto3
import json
import os
from datetime import datetime, timedelta
 
dynamodb = boto3.resource("dynamodb")
s3 = boto3.client("s3")
ses = boto3.client("ses")
 
orders_table = dynamodb.Table(os.environ["ORDERS_TABLE"])
REPORT_BUCKET = os.environ["REPORT_BUCKET"]
ADMIN_EMAIL = os.environ["ADMIN_EMAIL"]
 
def lambda_handler(event, context):
    """毎日 AM 9:00 (JST) に前日の注文レポートを生成"""
    yesterday = (datetime.utcnow() - timedelta(days=1)).strftime("%Y-%m-%d")
 
    # 前日の注文データを取得
    response = orders_table.query(
        IndexName="date-index",
        KeyConditionExpression="orderDate = :date",
        ExpressionAttributeValues={":date": yesterday}
    )
    orders = response["Items"]
 
    # レポート生成
    report = generate_report(yesterday, orders)
 
    # S3 にレポートを保存
    report_key = f"reports/daily/{yesterday}.json"
    s3.put_object(
        Bucket=REPORT_BUCKET,
        Key=report_key,
        Body=json.dumps(report, ensure_ascii=False, indent=2),
        ContentType="application/json"
    )
 
    # メール通知
    send_report_email(yesterday, report)
 
    return {"date": yesterday, "orderCount": len(orders)}
 
def generate_report(date, orders):
    total_revenue = sum(float(o.get("amount", 0)) for o in orders)
    categories = {}
    for order in orders:
        cat = order.get("category", "other")
        categories[cat] = categories.get(cat, 0) + 1
 
    return {
        "date": date,
        "totalOrders": len(orders),
        "totalRevenue": total_revenue,
        "averageOrderValue": total_revenue / len(orders) if orders else 0,
        "ordersByCategory": categories,
        "generatedAt": datetime.utcnow().isoformat()
    }
 
def send_report_email(date, report):
    ses.send_email(
        Source=ADMIN_EMAIL,
        Destination={"ToAddresses": [ADMIN_EMAIL]},
        Message={
            "Subject": {"Data": f"日次注文レポート: {date}"},
            "Body": {
                "Text": {
                    "Data": f"注文件数: {report['totalOrders']}\n"
                            f"売上合計: ¥{report['totalRevenue']:,.0f}\n"
                            f"平均注文額: ¥{report['averageOrderValue']:,.0f}"
                }
            }
        }
    )

# スケジュール駆動の SAM テンプレート
Resources:
  DailyReportFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: scheduled_report.lambda_handler
      Runtime: python3.12
      MemorySize: 256
      Timeout: 300
      Events:
        DailySchedule:
          Type: ScheduleV2
          Properties:
            ScheduleExpression: cron(0 0 * * ? *)  # 毎日 AM 9:00 JST (UTC 0:00)
            ScheduleExpressionTimezone: Asia/Tokyo
            RetryPolicy:
              MaximumRetryAttempts: 2
              MaximumEventAgeInSeconds: 3600

# フルスタックサーバーレスの SAM テンプレート
Resources:
  # S3 バケット (フロントエンド)
  WebBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${AWS::StackName}-web'
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
 
  # CloudFront OAC
  CloudFrontOAC:
    Type: AWS::CloudFront::OriginAccessControl
    Properties:
      OriginAccessControlConfig:
        Name: !Sub '${AWS::StackName}-oac'
        OriginAccessControlOriginType: s3
        SigningBehavior: always
        SigningProtocol: sigv4
 
  # CloudFront ディストリビューション
  Distribution:
    Type: AWS::CloudFront::Distribution
    Properties:
      DistributionConfig:
        Origins:
          - Id: S3Origin
            DomainName: !GetAtt WebBucket.RegionalDomainName
            OriginAccessControlId: !GetAtt CloudFrontOAC.Id
            S3OriginConfig:
              OriginAccessIdentity: ''
          - Id: ApiOrigin
            DomainName: !Sub '${HttpApi}.execute-api.${AWS::Region}.amazonaws.com'
            CustomOriginConfig:
              HTTPSPort: 443
              OriginProtocolPolicy: https-only
        DefaultCacheBehavior:
          TargetOriginId: S3Origin
          ViewerProtocolPolicy: redirect-to-https
          CachePolicyId: 658327ea-f89d-4fab-a63d-7e88639e58f6  # CachingOptimized
        CacheBehaviors:
          - PathPattern: /api/*
            TargetOriginId: ApiOrigin
            ViewerProtocolPolicy: https-only
            CachePolicyId: 4135ea2d-6df8-44a3-9df3-4b5a84be39ad  # CachingDisabled
            OriginRequestPolicyId: b689b0a8-53d0-40ab-baf2-68738e2966ac  # AllViewerExceptHostHeader
        DefaultRootObject: index.html
        CustomErrorResponses:
          - ErrorCode: 404
            ResponseCode: 200
            ResponsePagePath: /index.html  # SPA ルーティング対応
        Enabled: true
        HttpVersion: http2and3

# Provisioned Concurrency の設定
aws lambda put-provisioned-concurrency-config \
  --function-name my-api-function \
  --qualifier prod \
  --provisioned-concurrent-executions 10
 
# Application 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 5 \
  --max-capacity 50
 
aws application-autoscaling put-scaling-policy \
  --service-namespace lambda \
  --resource-id "function:my-api-function:prod" \
  --scalable-dimension "lambda:function:ProvisionedConcurrency" \
  --policy-name "target-tracking" \
  --policy-type TargetTrackingScaling \
  --target-tracking-scaling-policy-configuration '{
    "TargetValue": 0.7,
    "PredefinedMetricSpecification": {
      "PredefinedMetricType": "LambdaProvisionedConcurrencyUtilization"
    }
  }'

# SnapStart 対応 Lambda (Java)
Resources:
  JavaFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: com.example.Handler::handleRequest
      Runtime: java21
      MemorySize: 1024
      SnapStart:
        ApplyOn: PublishedVersions
      AutoPublishAlias: live

# [悪い例] 全件スキャンで検索
def search_items(keyword):
    result = table.scan(
        FilterExpression=Attr("name").contains(keyword)
    )
    return result["Items"]  # テーブル全体を読み取る
 
# [良い例] GSI を活用した効率的なクエリ
def search_items(category, date_from):
    result = table.query(
        IndexName="category-date-index",
        KeyConditionExpression=Key("category").eq(category) & Key("created_at").gte(date_from)
    )
    return result["Items"]  # 必要な範囲のみ読み取る

# [悪い例] 接続先やシークレットをハードコーディング
import boto3
 
def lambda_handler(event, context):
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table("my-production-table")  # テーブル名がハードコード
    api_key = "sk-abc123def456"  # シークレットがコード内
 
# [良い例] 環境変数と Secrets Manager を活用
import boto3
import os
import json
 
# ハンドラー外で初期化（コールドスタート時のみ実行）
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(os.environ["TABLE_NAME"])
secrets_client = boto3.client("secretsmanager")
 
_cached_secret = None
 
def get_api_key():
    global _cached_secret
    if _cached_secret is None:
        response = secrets_client.get_secret_value(
            SecretId=os.environ["API_KEY_SECRET_ARN"]
        )
        _cached_secret = json.loads(response["SecretString"])["api_key"]
    return _cached_secret
 
def lambda_handler(event, context):
    api_key = get_api_key()
    # 処理...

# 監視アラームの CloudFormation テンプレート
Resources:
  LambdaErrorAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub '${AWS::StackName}-lambda-errors'
      AlarmDescription: Lambda エラー率が5%を超過
      Namespace: AWS/Lambda
      MetricName: Errors
      Dimensions:
        - Name: FunctionName
          Value: !Ref ApiFunction
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 2
      Threshold: 5
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic
 
  ApiLatencyAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub '${AWS::StackName}-api-latency'
      AlarmDescription: API レイテンシ P99 が 3秒を超過
      Namespace: AWS/ApiGateway
      MetricName: Latency
      Dimensions:
        - Name: ApiName
          Value: !Ref HttpApi
      ExtendedStatistic: p99
      Period: 300
      EvaluationPeriods: 3
      Threshold: 3000
      ComparisonOperator: GreaterThanThreshold
      AlarmActions:
        - !Ref AlertTopic
 
  DLQAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub '${AWS::StackName}-dlq-messages'
      AlarmDescription: DLQ にメッセージが滞留
      Namespace: AWS/SQS
      MetricName: ApproximateNumberOfMessagesVisible
      Dimensions:
        - Name: QueueName
          Value: !GetAtt DeadLetterQueue.QueueName
      Statistic: Sum
      Period: 60
      EvaluationPeriods: 1
      Threshold: 1
      ComparisonOperator: GreaterThanOrEqualToThreshold
      AlarmActions:
        - !Ref AlertTopic
 
  AlertTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: !Sub '${AWS::StackName}-alerts'
      Subscription:
        - Endpoint: ops-team@example.com
          Protocol: email