# バージョニングを有効化
aws s3api put-bucket-versioning \
  --bucket my-app-bucket \
  --versioning-configuration Status=Enabled
 
# バージョニングの状態を確認
aws s3api get-bucket-versioning \
  --bucket my-app-bucket
 
# バージョン一覧の確認
aws s3api list-object-versions \
  --bucket my-app-bucket \
  --prefix report.pdf \
  --query '{Versions: Versions[].[Key,VersionId,LastModified,IsLatest], DeleteMarkers: DeleteMarkers[].[Key,VersionId]}'
 
# 特定バージョンの取得
aws s3api get-object \
  --bucket my-app-bucket \
  --key report.pdf \
  --version-id aaa111 \
  ./report-v1.pdf
 
# 削除マーカーを削除して復元
aws s3api delete-object \
  --bucket my-app-bucket \
  --key report.pdf \
  --version-id "DELETE_MARKER_VERSION_ID"
 
# 特定バージョンを完全削除
aws s3api delete-object \
  --bucket my-app-bucket \
  --key report.pdf \
  --version-id aaa111
 
# バージョニングの一時停止
aws s3api put-bucket-versioning \
  --bucket my-app-bucket \
  --versioning-configuration Status=Suspended

import boto3
from datetime import datetime, timezone
 
s3 = boto3.client('s3')
 
def list_versions(bucket, key):
    """オブジェクトの全バージョンを一覧表示"""
    response = s3.list_object_versions(Bucket=bucket, Prefix=key)
    for version in response.get('Versions', []):
        print(f"Version: {version['VersionId']} | "
              f"Size: {version['Size']} | "
              f"Date: {version['LastModified']} | "
              f"Latest: {version['IsLatest']}")
    for marker in response.get('DeleteMarkers', []):
        print(f"Delete Marker: {marker['VersionId']} | "
              f"Date: {marker['LastModified']}")
 
def restore_version(bucket, key, version_id):
    """特定バージョンを復元（コピーして最新に）"""
    s3.copy_object(
        Bucket=bucket,
        Key=key,
        CopySource={'Bucket': bucket, 'Key': key, 'VersionId': version_id}
    )
    print(f"Restored {key} to version {version_id}")
 
def delete_all_versions(bucket, key):
    """オブジェクトの全バージョンと削除マーカーを完全削除"""
    response = s3.list_object_versions(Bucket=bucket, Prefix=key)
 
    # バージョンを削除
    for version in response.get('Versions', []):
        s3.delete_object(
            Bucket=bucket,
            Key=key,
            VersionId=version['VersionId']
        )
        print(f"Deleted version: {version['VersionId']}")
 
    # 削除マーカーを削除
    for marker in response.get('DeleteMarkers', []):
        s3.delete_object(
            Bucket=bucket,
            Key=key,
            VersionId=marker['VersionId']
        )
        print(f"Deleted marker: {marker['VersionId']}")
 
def get_version_at_time(bucket, key, target_time):
    """指定時刻のバージョンを取得"""
    response = s3.list_object_versions(Bucket=bucket, Prefix=key)
    versions = sorted(
        response.get('Versions', []),
        key=lambda x: x['LastModified'],
        reverse=True
    )
    for v in versions:
        if v['LastModified'] <= target_time:
            return v
    return None

# 旧バージョンを90日後に Glacier に移行、365日後に削除
aws s3api put-bucket-lifecycle-configuration \
  --bucket my-app-bucket \
  --lifecycle-configuration '{
    "Rules": [
      {
        "ID": "ManageOldVersions",
        "Status": "Enabled",
        "Filter": {"Prefix": ""},
        "NoncurrentVersionTransitions": [
          {
            "NoncurrentDays": 90,
            "StorageClass": "GLACIER"
          },
          {
            "NoncurrentDays": 180,
            "StorageClass": "DEEP_ARCHIVE"
          }
        ],
        "NoncurrentVersionExpiration": {
          "NoncurrentDays": 365,
          "NewerNoncurrentVersions": 5
        }
      },
      {
        "ID": "CleanupDeleteMarkers",
        "Status": "Enabled",
        "Filter": {"Prefix": ""},
        "Expiration": {
          "ExpiredObjectDeleteMarker": true
        }
      },
      {
        "ID": "AbortIncompleteMultipart",
        "Status": "Enabled",
        "Filter": {"Prefix": ""},
        "AbortIncompleteMultipartUpload": {
          "DaysAfterInitiation": 7
        }
      }
    ]
  }'

AWSTemplateFormatVersion: '2010-09-09'
Description: S3 Bucket with Versioning and Lifecycle
 
Resources:
  AppBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Retain
    Properties:
      BucketName: !Sub '${AWS::StackName}-app-data'
      VersioningConfiguration:
        Status: Enabled
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !Ref BucketKmsKey
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      LifecycleConfiguration:
        Rules:
          - Id: ManageVersions
            Status: Enabled
            NoncurrentVersionTransitions:
              - StorageClass: GLACIER
                TransitionInDays: 90
            NoncurrentVersionExpiration:
              NoncurrentDays: 365
              NewerNoncurrentVersions: 3
          - Id: CleanupIncomplete
            Status: Enabled
            AbortIncompleteMultipartUpload:
              DaysAfterInitiation: 7
 
  BucketKmsKey:
    Type: AWS::KMS::Key
    Properties:
      Description: KMS key for S3 bucket encryption
      EnableKeyRotation: true
      KeyPolicy:
        Version: '2012-10-17'
        Statement:
          - Sid: AllowRootAccount
            Effect: Allow
            Principal:
              AWS: !Sub 'arn:aws:iam::${AWS::AccountId}:root'
            Action: 'kms:*'
            Resource: '*'

# 前提: 両バケットでバージョニングが有効であること
 
# レプリケーション用 IAM ロールを作成
cat > replication-trust.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "s3.amazonaws.com"},
    "Action": "sts:AssumeRole"
  }]
}
EOF
 
aws iam create-role \
  --role-name S3ReplicationRole \
  --assume-role-policy-document file://replication-trust.json
 
# レプリケーション用 IAM ポリシーを作成
cat > replication-policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetReplicationConfiguration",
        "s3:ListBucket"
      ],
      "Resource": "arn:aws:s3:::source-bucket"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObjectVersionForReplication",
        "s3:GetObjectVersionAcl",
        "s3:GetObjectVersionTagging"
      ],
      "Resource": "arn:aws:s3:::source-bucket/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:ReplicateObject",
        "s3:ReplicateDelete",
        "s3:ReplicateTags"
      ],
      "Resource": "arn:aws:s3:::destination-bucket/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "kms:Decrypt"
      ],
      "Resource": "arn:aws:kms:ap-northeast-1:123456789012:key/source-key-id",
      "Condition": {
        "StringLike": {
          "kms:ViaService": "s3.ap-northeast-1.amazonaws.com"
        }
      }
    },
    {
      "Effect": "Allow",
      "Action": [
        "kms:Encrypt"
      ],
      "Resource": "arn:aws:kms:ap-northeast-3:123456789012:key/dest-key-id",
      "Condition": {
        "StringLike": {
          "kms:ViaService": "s3.ap-northeast-3.amazonaws.com"
        }
      }
    }
  ]
}
EOF
 
aws iam put-role-policy \
  --role-name S3ReplicationRole \
  --policy-name S3ReplicationPolicy \
  --policy-document file://replication-policy.json
 
# レプリケーションルールを設定
aws s3api put-bucket-replication \
  --bucket source-bucket \
  --replication-configuration '{
  "Role": "arn:aws:iam::123456789012:role/S3ReplicationRole",
  "Rules": [
    {
      "ID": "ReplicateAll",
      "Status": "Enabled",
      "Priority": 1,
      "Filter": {"Prefix": ""},
      "Destination": {
        "Bucket": "arn:aws:s3:::destination-bucket",
        "StorageClass": "STANDARD_IA",
        "ReplicationTime": {
          "Status": "Enabled",
          "Time": {"Minutes": 15}
        },
        "Metrics": {
          "Status": "Enabled",
          "EventThreshold": {"Minutes": 15}
        },
        "EncryptionConfiguration": {
          "ReplicaKmsKeyID": "arn:aws:kms:ap-northeast-3:123456789012:key/dest-key-id"
        }
      },
      "SourceSelectionCriteria": {
        "SseKmsEncryptedObjects": {
          "Status": "Enabled"
        }
      },
      "DeleteMarkerReplication": {"Status": "Enabled"}
    }
  ]
}'
 
# レプリケーションの状態確認
aws s3api get-bucket-replication --bucket source-bucket
 
# レプリケーションメトリクスの確認（S3 Replication Time Control）
aws cloudwatch get-metric-statistics \
  --namespace AWS/S3 \
  --metric-name ReplicationLatency \
  --dimensions Name=SourceBucket,Value=source-bucket \
              Name=DestinationBucket,Value=destination-bucket \
              Name=RuleId,Value=ReplicateAll \
  --start-time 2026-02-14T00:00:00Z \
  --end-time 2026-02-15T00:00:00Z \
  --period 3600 \
  --statistics Average

# S3 インベントリレポートの設定（バッチレプリケーションの入力）
aws s3api put-bucket-inventory-configuration \
  --bucket source-bucket \
  --id weekly-inventory \
  --inventory-configuration '{
    "Destination": {
      "S3BucketDestination": {
        "AccountId": "123456789012",
        "Bucket": "arn:aws:s3:::inventory-reports-bucket",
        "Format": "CSV",
        "Prefix": "inventory"
      }
    },
    "IsEnabled": true,
    "Id": "weekly-inventory",
    "IncludedObjectVersions": "All",
    "Schedule": {"Frequency": "Weekly"},
    "OptionalFields": [
      "Size", "LastModifiedDate", "StorageClass",
      "ReplicationStatus", "EncryptionStatus"
    ]
  }'
 
# バッチレプリケーションジョブの作成
aws s3control create-job \
  --account-id 123456789012 \
  --operation '{"S3ReplicateObject":{}}' \
  --manifest '{
    "Spec": {
      "Format": "S3InventoryReport_CSV_20211130",
      "Fields": ["Bucket","Key","VersionId"]
    },
    "Location": {
      "ObjectArn": "arn:aws:s3:::inventory-reports-bucket/inventory/source-bucket/weekly-inventory/data/manifest.json",
      "ETag": "abc123"
    }
  }' \
  --report '{
    "Bucket": "arn:aws:s3:::batch-reports-bucket",
    "Format": "Report_CSV_20180820",
    "Enabled": true,
    "Prefix": "batch-replication",
    "ReportScope": "AllTasks"
  }' \
  --priority 42 \
  --role-arn arn:aws:iam::123456789012:role/S3BatchRole \
  --confirmation-required

import boto3
import json
 
s3 = boto3.client('s3')
 
# CSV ファイルから特定条件のデータを抽出
def query_csv(bucket, key, sql):
    response = s3.select_object_content(
        Bucket=bucket,
        Key=key,
        ExpressionType='SQL',
        Expression=sql,
        InputSerialization={
            'CSV': {
                'FileHeaderInfo': 'USE',
                'RecordDelimiter': '\n',
                'FieldDelimiter': ','
            },
            'CompressionType': 'GZIP'
        },
        OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
    )
 
    records = []
    for event in response['Payload']:
        if 'Records' in event:
            records.append(event['Records']['Payload'].decode('utf-8'))
        elif 'Stats' in event:
            stats = event['Stats']['Details']
            print(f"Scanned: {stats['BytesScanned']} bytes, "
                  f"Processed: {stats['BytesProcessed']} bytes, "
                  f"Returned: {stats['BytesReturned']} bytes")
    return records
 
# 使用例: 2024年の東京のデータのみ取得
results = query_csv(
    'analytics-bucket',
    'data/sales-2024.csv.gz',
    "SELECT s.date, s.product, s.amount FROM s3object s WHERE s.region = 'Tokyo' AND CAST(s.amount AS INT) > 10000"
)
 
# JSON ファイルからクエリ
response = s3.select_object_content(
    Bucket='data-bucket',
    Key='logs/events.json',
    ExpressionType='SQL',
    Expression="SELECT s.timestamp, s.level, s.message FROM s3object s WHERE s.level = 'ERROR'",
    InputSerialization={'JSON': {'Type': 'LINES'}},
    OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
)
 
# Parquet ファイルからクエリ（列指向のため、列選択で最大の効果）
response = s3.select_object_content(
    Bucket='analytics-bucket',
    Key='data/events.parquet',
    ExpressionType='SQL',
    Expression="SELECT event_type, COUNT(*) as cnt FROM s3object s GROUP BY event_type",
    InputSerialization={'Parquet': {}},
    OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
)

import boto3
import json
from typing import List, Dict, Any, Optional
 
class S3SelectQuery:
    """S3 Select のラッパークラス"""
 
    def __init__(self, region_name='ap-northeast-1'):
        self.s3 = boto3.client('s3', region_name=region_name)
 
    def query_csv(
        self,
        bucket: str,
        key: str,
        sql: str,
        compression: str = 'NONE',
        delimiter: str = ',',
        header: str = 'USE'
    ) -> List[Dict[str, Any]]:
        """CSV ファイルに対して S3 Select クエリを実行"""
        response = self.s3.select_object_content(
            Bucket=bucket,
            Key=key,
            ExpressionType='SQL',
            Expression=sql,
            InputSerialization={
                'CSV': {
                    'FileHeaderInfo': header,
                    'RecordDelimiter': '\n',
                    'FieldDelimiter': delimiter
                },
                'CompressionType': compression
            },
            OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
        )
        return self._parse_response(response)
 
    def query_json(
        self,
        bucket: str,
        key: str,
        sql: str,
        json_type: str = 'LINES',
        compression: str = 'NONE'
    ) -> List[Dict[str, Any]]:
        """JSON ファイルに対して S3 Select クエリを実行"""
        response = self.s3.select_object_content(
            Bucket=bucket,
            Key=key,
            ExpressionType='SQL',
            Expression=sql,
            InputSerialization={
                'JSON': {'Type': json_type},
                'CompressionType': compression
            },
            OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
        )
        return self._parse_response(response)
 
    def query_parquet(
        self,
        bucket: str,
        key: str,
        sql: str
    ) -> List[Dict[str, Any]]:
        """Parquet ファイルに対して S3 Select クエリを実行"""
        response = self.s3.select_object_content(
            Bucket=bucket,
            Key=key,
            ExpressionType='SQL',
            Expression=sql,
            InputSerialization={'Parquet': {}},
            OutputSerialization={'JSON': {'RecordDelimiter': '\n'}}
        )
        return self._parse_response(response)
 
    def _parse_response(self, response) -> List[Dict[str, Any]]:
        """レスポンスをパースして辞書のリストとして返す"""
        records = []
        stats = {}
        for event in response['Payload']:
            if 'Records' in event:
                payload = event['Records']['Payload'].decode('utf-8')
                for line in payload.strip().split('\n'):
                    if line:
                        records.append(json.loads(line))
            elif 'Stats' in event:
                details = event['Stats']['Details']
                stats = {
                    'bytes_scanned': details['BytesScanned'],
                    'bytes_processed': details['BytesProcessed'],
                    'bytes_returned': details['BytesReturned'],
                    'compression_ratio': (
                        1 - details['BytesReturned'] / details['BytesScanned']
                    ) if details['BytesScanned'] > 0 else 0
                }
                print(f"S3 Select Stats: scanned={stats['bytes_scanned']}, "
                      f"returned={stats['bytes_returned']}, "
                      f"compression={stats['compression_ratio']:.1%}")
        return records
 
# 使用例
sq = S3SelectQuery()
 
# 大容量 CSV からの高速フィルタリング
errors = sq.query_csv(
    'log-bucket',
    'access-logs/2026/02/access.csv.gz',
    "SELECT s.timestamp, s.status, s.path FROM s3object s WHERE CAST(s.status AS INT) >= 500",
    compression='GZIP'
)
 
# JSON Lines ログからの抽出
slow_queries = sq.query_json(
    'app-logs',
    'db-logs/slow-queries.json',
    "SELECT s.query, s.duration, s.timestamp FROM s3object s WHERE s.duration > 1000"
)

# Transfer Acceleration を有効化
aws s3api put-bucket-accelerate-configuration \
  --bucket my-global-bucket \
  --accelerate-configuration Status=Enabled
 
# 有効化の確認
aws s3api get-bucket-accelerate-configuration \
  --bucket my-global-bucket
 
# Acceleration エンドポイントでアップロード
aws s3 cp large-file.zip \
  s3://my-global-bucket/uploads/ \
  --endpoint-url https://my-global-bucket.s3-accelerate.amazonaws.com
 
# マルチパートアップロードと組み合わせ（大容量ファイル）
aws s3 cp large-dataset.tar.gz \
  s3://my-global-bucket/datasets/ \
  --endpoint-url https://my-global-bucket.s3-accelerate.amazonaws.com \
  --expected-size 10737418240
 
# 速度比較ツール
# https://s3-accelerate-speedtest.s3-accelerate.amazonaws.com/en/accelerate-speed-comparsion.html

import boto3
from boto3.s3.transfer import TransferConfig
import time
 
# Acceleration エンドポイントを使用
s3 = boto3.client(
    's3',
    config=boto3.session.Config(s3={'use_accelerate_endpoint': True})
)
 
# マルチパート設定（大容量ファイル向け）
config = TransferConfig(
    multipart_threshold=100 * 1024 * 1024,   # 100MB
    max_concurrency=10,
    multipart_chunksize=100 * 1024 * 1024,
    use_threads=True
)
 
# プログレスコールバック付きアップロード
class ProgressTracker:
    def __init__(self, filename, filesize):
        self.filename = filename
        self.filesize = filesize
        self.uploaded = 0
        self.start_time = time.time()
 
    def __call__(self, bytes_amount):
        self.uploaded += bytes_amount
        percentage = (self.uploaded / self.filesize) * 100
        elapsed = time.time() - self.start_time
        speed = self.uploaded / elapsed / 1024 / 1024 if elapsed > 0 else 0
        print(f"\r{self.filename}: {percentage:.1f}% ({speed:.1f} MB/s)", end='')
 
import os
filepath = 'large-dataset.tar.gz'
filesize = os.path.getsize(filepath)
progress = ProgressTracker(filepath, filesize)
 
s3.upload_file(
    filepath,
    'my-global-bucket',
    'datasets/large-dataset.tar.gz',
    Config=config,
    Callback=progress
)
print(f"\nUpload complete in {time.time() - progress.start_time:.1f}s")

import boto3
import time
import os
 
def benchmark_transfer(bucket, key, filepath, use_acceleration=False):
    """通常転送と Acceleration 転送の速度を比較"""
    config_kwargs = {}
    if use_acceleration:
        config_kwargs['config'] = boto3.session.Config(
            s3={'use_accelerate_endpoint': True}
        )
 
    s3 = boto3.client('s3', **config_kwargs)
    filesize = os.path.getsize(filepath)
 
    start = time.time()
    s3.upload_file(filepath, bucket, key)
    elapsed = time.time() - start
 
    speed_mbps = (filesize / 1024 / 1024) / elapsed
    print(f"{'Accelerated' if use_acceleration else 'Standard'}: "
          f"{elapsed:.2f}s ({speed_mbps:.2f} MB/s)")
    return elapsed
 
# ベンチマーク実行
test_file = 'test-100mb.bin'
print("Transfer speed comparison:")
standard_time = benchmark_transfer('my-bucket', 'test/std', test_file, False)
accel_time = benchmark_transfer('my-bucket', 'test/accel', test_file, True)
improvement = ((standard_time - accel_time) / standard_time) * 100
print(f"Improvement: {improvement:.1f}%")

# Intelligent-Tiering アーカイブアクセス層を有効化
aws s3api put-bucket-intelligent-tiering-configuration \
  --bucket my-data-bucket \
  --id archive-config \
  --intelligent-tiering-configuration '{
    "Id": "archive-config",
    "Status": "Enabled",
    "Tierings": [
      {
        "AccessTier": "ARCHIVE_ACCESS",
        "Days": 90
      },
      {
        "AccessTier": "DEEP_ARCHIVE_ACCESS",
        "Days": 180
      }
    ],
    "Filter": {
      "Prefix": "data/"
    }
  }'
 
# ライフサイクルで Intelligent-Tiering に自動移行
aws s3api put-bucket-lifecycle-configuration \
  --bucket my-data-bucket \
  --lifecycle-configuration '{
    "Rules": [
      {
        "ID": "MoveToIT",
        "Status": "Enabled",
        "Filter": {"Prefix": "data/"},
        "Transitions": [
          {
            "Days": 0,
            "StorageClass": "INTELLIGENT_TIERING"
          }
        ]
      }
    ]
  }'

# Storage Lens 設定を作成
aws s3control put-storage-lens-configuration \
  --account-id 123456789012 \
  --config-id my-storage-lens \
  --storage-lens-configuration '{
    "Id": "my-storage-lens",
    "AccountLevel": {
      "BucketLevel": {
        "ActivityMetrics": {"IsEnabled": true},
        "AdvancedCostOptimizationMetrics": {"IsEnabled": true},
        "AdvancedDataProtectionMetrics": {"IsEnabled": true},
        "DetailedStatusCodesMetrics": {"IsEnabled": true},
        "PrefixLevel": {
          "StorageMetrics": {
            "IsEnabled": true,
            "SelectionCriteria": {
              "MaxDepth": 3,
              "MinStorageBytesPercentage": 1.0
            }
          }
        }
      }
    },
    "DataExport": {
      "S3BucketDestination": {
        "AccountId": "123456789012",
        "Arn": "arn:aws:s3:::storage-lens-reports",
        "Format": "CSV",
        "OutputSchemaVersion": "V_1",
        "Encryption": {
          "SSES3": {}
        }
      },
      "CloudWatchMetrics": {
        "IsEnabled": true
      }
    },
    "IsEnabled": true
  }'

# S3 用 Gateway エンドポイント（無料）
aws ec2 create-vpc-endpoint \
  --vpc-id vpc-xxx \
  --service-name com.amazonaws.ap-northeast-1.s3 \
  --route-table-ids rtb-xxx \
  --vpc-endpoint-type Gateway
 
# エンドポイントポリシーでアクセスを制限
aws ec2 modify-vpc-endpoint \
  --vpc-endpoint-id vpce-xxx \
  --policy-document '{
    "Version": "2012-10-17",
    "Statement": [
      {
        "Sid": "AllowSpecificBuckets",
        "Effect": "Allow",
        "Principal": "*",
        "Action": [
          "s3:GetObject",
          "s3:PutObject",
          "s3:ListBucket"
        ],
        "Resource": [
          "arn:aws:s3:::my-app-bucket",
          "arn:aws:s3:::my-app-bucket/*",
          "arn:aws:s3:::my-logs-bucket",
          "arn:aws:s3:::my-logs-bucket/*"
        ]
      }
    ]
  }'
 
# エンドポイント経由のアクセスは NAT Gateway の料金が不要
# → データ転送コストを大幅に削減

import boto3
from datetime import datetime, timedelta
 
def analyze_s3_costs(bucket_name):
    """S3 バケットのコスト最適化レポートを生成"""
    s3 = boto3.client('s3')
    cloudwatch = boto3.client('cloudwatch')
 
    # バケットサイズの取得
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=1)
 
    size_response = cloudwatch.get_metric_statistics(
        Namespace='AWS/S3',
        MetricName='BucketSizeBytes',
        Dimensions=[
            {'Name': 'BucketName', 'Value': bucket_name},
            {'Name': 'StorageType', 'Value': 'StandardStorage'}
        ],
        StartTime=start_time,
        EndTime=end_time,
        Period=86400,
        Statistics=['Average']
    )
 
    # オブジェクト数の取得
    count_response = cloudwatch.get_metric_statistics(
        Namespace='AWS/S3',
        MetricName='NumberOfObjects',
        Dimensions=[
            {'Name': 'BucketName', 'Value': bucket_name},
            {'Name': 'StorageType', 'Value': 'AllStorageTypes'}
        ],
        StartTime=start_time,
        EndTime=end_time,
        Period=86400,
        Statistics=['Average']
    )
 
    # ストレージクラス別の分析
    paginator = s3.get_paginator('list_objects_v2')
    storage_classes = {}
    total_size = 0
    for page in paginator.paginate(Bucket=bucket_name, MaxKeys=1000):
        for obj in page.get('Contents', []):
            sc = obj.get('StorageClass', 'STANDARD')
            if sc not in storage_classes:
                storage_classes[sc] = {'count': 0, 'size': 0}
            storage_classes[sc]['count'] += 1
            storage_classes[sc]['size'] += obj['Size']
            total_size += obj['Size']
 
    # レポート出力
    print(f"=== S3 Cost Analysis: {bucket_name} ===")
    print(f"Total Size: {total_size / 1024 / 1024 / 1024:.2f} GB")
    print(f"\nStorage Class Distribution:")
    for sc, info in storage_classes.items():
        pct = (info['size'] / total_size * 100) if total_size > 0 else 0
        print(f"  {sc}: {info['count']} objects, "
              f"{info['size'] / 1024 / 1024:.1f} MB ({pct:.1f}%)")
 
    # 最適化の推奨
    print(f"\n=== Recommendations ===")
    std_size = storage_classes.get('STANDARD', {}).get('size', 0)
    if std_size > 100 * 1024 * 1024 * 1024:  # 100GB以上
        print("- Consider Intelligent-Tiering for large STANDARD storage")
    if 'STANDARD' in storage_classes and storage_classes['STANDARD']['count'] > 10000:
        print("- Enable S3 Inventory for detailed analysis")
    print("- Check for incomplete multipart uploads")
    print("- Review lifecycle rules for old objects")
 
# 実行
analyze_s3_costs('my-app-bucket')

# Lambda へのイベント通知
aws s3api put-bucket-notification-configuration \
  --bucket my-app-bucket \
  --notification-configuration '{
  "LambdaFunctionConfigurations": [
    {
      "Id": "ProcessUploadedImages",
      "LambdaFunctionArn": "arn:aws:lambda:ap-northeast-1:123456789012:function:image-processor",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {"Name": "prefix", "Value": "uploads/images/"},
            {"Name": "suffix", "Value": ".jpg"}
          ]
        }
      }
    },
    {
      "Id": "ProcessUploadedPNG",
      "LambdaFunctionArn": "arn:aws:lambda:ap-northeast-1:123456789012:function:image-processor",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {
        "Key": {
          "FilterRules": [
            {"Name": "prefix", "Value": "uploads/images/"},
            {"Name": "suffix", "Value": ".png"}
          ]
        }
      }
    }
  ],
  "QueueConfigurations": [
    {
      "Id": "LogDeletion",
      "QueueArn": "arn:aws:sqs:ap-northeast-1:123456789012:s3-deletion-queue",
      "Events": ["s3:ObjectRemoved:*"]
    }
  ],
  "EventBridgeConfiguration": {}
}'

# Lambda: S3 イベントを処理する関数
import boto3
import json
import urllib.parse
 
s3 = boto3.client('s3')
 
def handler(event, context):
    """S3 イベント通知を処理する Lambda ハンドラ"""
 
    # S3 イベント通知の場合
    if 'Records' in event:
        for record in event['Records']:
            bucket = record['s3']['bucket']['name']
            key = urllib.parse.unquote_plus(
                record['s3']['object']['key'], encoding='utf-8'
            )
            size = record['s3']['object'].get('size', 0)
            event_name = record['eventName']
 
            print(f"Event: {event_name}, Bucket: {bucket}, "
                  f"Key: {key}, Size: {size}")
 
            if event_name.startswith('ObjectCreated:'):
                process_new_object(bucket, key, size)
            elif event_name.startswith('ObjectRemoved:'):
                handle_deletion(bucket, key)
 
    # EventBridge 経由の場合
    elif event.get('source') == 'aws.s3':
        detail = event['detail']
        bucket = detail['bucket']['name']
        key = detail['object']['key']
        process_new_object(bucket, key, detail['object']['size'])
 
    return {'statusCode': 200}
 
def process_new_object(bucket, key, size):
    """新規オブジェクトの処理"""
    if key.endswith(('.jpg', '.jpeg', '.png', '.webp')):
        # 画像処理パイプラインを起動
        generate_thumbnails(bucket, key)
    elif key.endswith('.csv'):
        # CSV データの ETL 処理
        trigger_etl_job(bucket, key)
    elif key.endswith('.mp4'):
        # 動画トランスコーディング
        start_transcode_job(bucket, key)
 
def handle_deletion(bucket, key):
    """オブジェクト削除時の処理"""
    print(f"Object deleted: s3://{bucket}/{key}")
    # CDN キャッシュの無効化
    # 検索インデックスからの削除
    # 関連リソースのクリーンアップ

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
 
Resources:
  UploadBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${AWS::StackName}-uploads'
      NotificationConfiguration:
        LambdaConfigurations:
          - Event: 's3:ObjectCreated:*'
            Function: !GetAtt ImageProcessorFunction.Arn
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: images/
        EventBridgeConfiguration:
          EventBridgeEnabled: true
 
  ImageProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      Runtime: python3.12
      Handler: index.handler
      MemorySize: 1024
      Timeout: 300
      Policies:
        - S3ReadPolicy:
            BucketName: !Sub '${AWS::StackName}-uploads'
        - S3CrudPolicy:
            BucketName: !Sub '${AWS::StackName}-processed'
 
  ImageProcessorPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref ImageProcessorFunction
      Action: lambda:InvokeFunction
      Principal: s3.amazonaws.com
      SourceArn: !GetAtt UploadBucket.Arn
 
  # EventBridge ルール（高度なフィルタリング）
  LargeFileRule:
    Type: AWS::Events::Rule
    Properties:
      EventPattern:
        source:
          - aws.s3
        detail-type:
          - 'Object Created'
        detail:
          bucket:
            name:
              - !Ref UploadBucket
          object:
            size:
              - numeric:
                  - '>='
                  - 104857600  # 100MB以上
      Targets:
        - Arn: !GetAtt LargeFileProcessorFunction.Arn
          Id: LargeFileTarget

# Object Lock 有効なバケットを作成（作成時のみ設定可能）
aws s3api create-bucket \
  --bucket compliance-bucket \
  --region ap-northeast-1 \
  --create-bucket-configuration LocationConstraint=ap-northeast-1 \
  --object-lock-enabled-for-bucket
 
# デフォルトの Object Lock 設定
aws s3api put-object-lock-configuration \
  --bucket compliance-bucket \
  --object-lock-configuration '{
    "ObjectLockEnabled": "Enabled",
    "Rule": {
      "DefaultRetention": {
        "Mode": "COMPLIANCE",
        "Days": 365
      }
    }
  }'
 
# 個別オブジェクトに Object Lock を設定
aws s3api put-object-retention \
  --bucket compliance-bucket \
  --key financial-reports/2025-annual.pdf \
  --retention '{
    "Mode": "COMPLIANCE",
    "RetainUntilDate": "2030-12-31T00:00:00Z"
  }'
 
# Legal Hold の設定
aws s3api put-object-legal-hold \
  --bucket compliance-bucket \
  --key contracts/nda-2025.pdf \
  --legal-hold '{"Status": "ON"}'
 
# Legal Hold の解除
aws s3api put-object-legal-hold \
  --bucket compliance-bucket \
  --key contracts/nda-2025.pdf \
  --legal-hold '{"Status": "OFF"}'

# ストレージクラスの一括変更ジョブ
aws s3control create-job \
  --account-id 123456789012 \
  --operation '{
    "S3PutObjectCopy": {
      "TargetResource": "arn:aws:s3:::my-app-bucket",
      "StorageClass": "INTELLIGENT_TIERING",
      "MetadataDirective": "COPY"
    }
  }' \
  --manifest '{
    "Spec": {
      "Format": "S3InventoryReport_CSV_20211130",
      "Fields": ["Bucket", "Key", "VersionId"]
    },
    "Location": {
      "ObjectArn": "arn:aws:s3:::inventory-bucket/manifest.json",
      "ETag": "abc123def456"
    }
  }' \
  --report '{
    "Bucket": "arn:aws:s3:::reports-bucket",
    "Format": "Report_CSV_20180820",
    "Enabled": true,
    "Prefix": "batch-reports/storage-class-change",
    "ReportScope": "FailedTasksOnly"
  }' \
  --priority 10 \
  --role-arn arn:aws:iam::123456789012:role/S3BatchRole \
  --confirmation-required
 
# タグの一括追加ジョブ
aws s3control create-job \
  --account-id 123456789012 \
  --operation '{
    "S3PutObjectTagging": {
      "TagSet": [
        {"Key": "Department", "Value": "Finance"},
        {"Key": "Classification", "Value": "Internal"}
      ]
    }
  }' \
  --manifest '{
    "Spec": {
      "Format": "S3BatchOperations_CSV_20180820",
      "Fields": ["Bucket", "Key"]
    },
    "Location": {
      "ObjectArn": "arn:aws:s3:::manifests/tag-objects.csv",
      "ETag": "xyz789"
    }
  }' \
  --report '{
    "Bucket": "arn:aws:s3:::reports-bucket",
    "Format": "Report_CSV_20180820",
    "Enabled": true,
    "ReportScope": "AllTasks"
  }' \
  --priority 20 \
  --role-arn arn:aws:iam::123456789012:role/S3BatchRole \
  --no-confirmation-required
 
# ジョブの状態確認
aws s3control describe-job \
  --account-id 123456789012 \
  --job-id "job-id-here"
 
# ジョブの一覧
aws s3control list-jobs \
  --account-id 123456789012 \
  --job-statuses Active Complete

# HTTPS 強制 + VPC エンドポイント制限 + 暗号化強制
aws s3api put-bucket-policy --bucket secure-bucket --policy '{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ForceHTTPS",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:*",
      "Resource": [
        "arn:aws:s3:::secure-bucket",
        "arn:aws:s3:::secure-bucket/*"
      ],
      "Condition": {
        "Bool": {
          "aws:SecureTransport": "false"
        }
      }
    },
    {
      "Sid": "RestrictToVPC",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:*",
      "Resource": [
        "arn:aws:s3:::secure-bucket",
        "arn:aws:s3:::secure-bucket/*"
      ],
      "Condition": {
        "StringNotEquals": {
          "aws:SourceVpce": "vpce-xxxxxxxx"
        }
      }
    },
    {
      "Sid": "ForceSSEKMS",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::secure-bucket/*",
      "Condition": {
        "StringNotEquals": {
          "s3:x-amz-server-side-encryption": "aws:kms"
        }
      }
    },
    {
      "Sid": "DenyUnencryptedTransport",
      "Effect": "Deny",
      "Principal": "*",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::secure-bucket/*",
      "Condition": {
        "Null": {
          "s3:x-amz-server-side-encryption": "true"
        }
      }
    }
  ]
}'
 
# S3 Access Point の作成（アクセス制御の簡素化）
aws s3control create-access-point \
  --account-id 123456789012 \
  --name app-readonly-ap \
  --bucket secure-bucket \
  --vpc-configuration VpcId=vpc-xxx \
  --public-access-block-configuration \
    BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true

# IAM Access Analyzer で外部アクセスを検出
aws accessanalyzer create-analyzer \
  --analyzer-name s3-analyzer \
  --type ACCOUNT
 
# 検出結果の確認
aws accessanalyzer list-findings \
  --analyzer-arn arn:aws:access-analyzer:ap-northeast-1:123456789012:analyzer/s3-analyzer \
  --filter '{
    "resourceType": {
      "eq": ["AWS::S3::Bucket"]
    },
    "status": {
      "eq": ["ACTIVE"]
    }
  }'

import boto3
from concurrent.futures import ThreadPoolExecutor, as_completed
import os
 
s3 = boto3.client('s3')
 
def parallel_upload(bucket, prefix, file_list, max_workers=20):
    """並列アップロードで高スループットを実現"""
    results = []
 
    def upload_one(filepath):
        key = f"{prefix}/{os.path.basename(filepath)}"
        s3.upload_file(filepath, bucket, key)
        return {'file': filepath, 'key': key, 'status': 'success'}
 
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(upload_one, f): f for f in file_list
        }
        for future in as_completed(futures):
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                results.append({
                    'file': futures[future],
                    'status': 'error',
                    'error': str(e)
                })
 
    success = sum(1 for r in results if r['status'] == 'success')
    print(f"Uploaded {success}/{len(file_list)} files")
    return results
 
def parallel_download(bucket, prefix, dest_dir, max_workers=20):
    """並列ダウンロードで高スループットを実現"""
    paginator = s3.get_paginator('list_objects_v2')
    keys = []
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            keys.append(obj['Key'])
 
    os.makedirs(dest_dir, exist_ok=True)
 
    def download_one(key):
        local_path = os.path.join(dest_dir, os.path.basename(key))
        s3.download_file(bucket, key, local_path)
        return {'key': key, 'local': local_path, 'status': 'success'}
 
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(download_one, k): k for k in keys
        }
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                results.append({
                    'key': futures[future],
                    'status': 'error',
                    'error': str(e)
                })
 
    success = sum(1 for r in results if r['status'] == 'success')
    print(f"Downloaded {success}/{len(keys)} files")
    return results

# S3 上のデータを Athena でクエリ
aws athena start-query-execution \
  --query-string "
    CREATE EXTERNAL TABLE IF NOT EXISTS access_logs (
      request_time string,
      remote_ip string,
      request_method string,
      request_path string,
      status_code int,
      bytes_sent bigint,
      user_agent string
    )
    ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
    WITH SERDEPROPERTIES (
      'input.regex' = '([^ ]*) ([^ ]*) ([^ ]*) ([^ ]*) ([0-9]*) ([0-9]*) (.*)$'
    )
    LOCATION 's3://my-logs-bucket/access-logs/'
  " \
  --result-configuration OutputLocation=s3://athena-results/

import * as cdk from 'aws-cdk-lib';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as s3n from 'aws-cdk-lib/aws-s3-notifications';
import * as iam from 'aws-cdk-lib/aws-iam';
import { Construct } from 'constructs';
 
export class S3AdvancedStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);
 
    // メインバケット（バージョニング + ライフサイクル）
    const mainBucket = new s3.Bucket(this, 'MainBucket', {
      bucketName: `${this.stackName}-main-data`,
      versioned: true,
      encryption: s3.BucketEncryption.KMS_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
      removalPolicy: cdk.RemovalPolicy.RETAIN,
      lifecycleRules: [
        {
          id: 'TransitionToIA',
          transitions: [
            {
              storageClass: s3.StorageClass.INFREQUENT_ACCESS,
              transitionAfter: cdk.Duration.days(30),
            },
            {
              storageClass: s3.StorageClass.INTELLIGENT_TIERING,
              transitionAfter: cdk.Duration.days(60),
            },
          ],
          noncurrentVersionTransitions: [
            {
              storageClass: s3.StorageClass.GLACIER,
              transitionAfter: cdk.Duration.days(90),
            },
          ],
          noncurrentVersionExpiration: cdk.Duration.days(365),
          abortIncompleteMultipartUploadAfter: cdk.Duration.days(7),
        },
      ],
      intelligentTieringConfigurations: [
        {
          name: 'archive-config',
          archiveAccessTierTime: cdk.Duration.days(90),
          deepArchiveAccessTierTime: cdk.Duration.days(180),
        },
      ],
    });
 
    // DR 用レプリカバケット
    const replicaBucket = new s3.Bucket(this, 'ReplicaBucket', {
      bucketName: `${this.stackName}-replica`,
      versioned: true,
      encryption: s3.BucketEncryption.KMS_MANAGED,
      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
      enforceSSL: true,
    });
 
    // イベント通知: 画像アップロード時に Lambda を実行
    const imageProcessor = new lambda.Function(this, 'ImageProcessor', {
      runtime: lambda.Runtime.PYTHON_3_12,
      handler: 'index.handler',
      code: lambda.Code.fromAsset('lambda/image-processor'),
      memorySize: 1024,
      timeout: cdk.Duration.minutes(5),
    });
 
    mainBucket.addEventNotification(
      s3.EventType.OBJECT_CREATED,
      new s3n.LambdaDestination(imageProcessor),
      { prefix: 'uploads/images/', suffix: '.jpg' }
    );
 
    mainBucket.grantRead(imageProcessor);
 
    // S3 Access Point
    new s3.CfnAccessPoint(this, 'ReadOnlyAccessPoint', {
      bucket: mainBucket.bucketName,
      name: 'readonly-ap',
      vpcConfiguration: {
        vpcId: 'vpc-xxx',
      },
      publicAccessBlockConfiguration: {
        blockPublicAcls: true,
        blockPublicPolicy: true,
        ignorePublicAcls: true,
        restrictPublicBuckets: true,
      },
    });
  }
}

{
  "Rules": [{
    "ID": "ExpireOldVersions",
    "Status": "Enabled",
    "Filter": {"Prefix": ""},
    "NoncurrentVersionTransitions": [
      {"NoncurrentDays": 30, "StorageClass": "STANDARD_IA"},
      {"NoncurrentDays": 90, "StorageClass": "GLACIER"}
    ],
    "NoncurrentVersionExpiration": {
      "NoncurrentDays": 365,
      "NewerNoncurrentVersions": 3
    }
  }]
}

# 全アカウントレベルでパブリックアクセスをブロック
aws s3control put-public-access-block \
  --account-id 123456789012 \
  --public-access-block-configuration \
    BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true